fix(thumbnails): a folder check answers for the folder as it is now (#320)

Probe.check() joined a check already running, or a finished one nobody had
collected. A folder deleted right after configure() woke the probe therefore
stayed available until the next probe, which failed the Overview notice spec in
the merge queue. It now waits for a running check and starts a new one, still
never two at once, and a stuck check still times out. The spec also asserts the
fixture's removal took effect.

TESTING.md

@@ -5723,6 +5723,73 @@ are closed before `os.replace`; renames stay inside one folder; folder identity
 folder are refused. The browser test server closes its own log handlers before its temp
 root (which now holds the logs) is removed.
 
+### Thumbnail location contract (#320)
+
+Previews live in each source's `thumbnails/` (the default) or in one folder, `<folder>/<source key>/…`. Every reader,
+writer and remover asks `py/api/preview_store.py`; `tests/test_preview_store.py` fails on a `thumbnails` path join
+anywhere else. The setting is `thumbnails.location` in config.json, and only the Move thumbnails job writes it: POST
+`/settings` answers 409 `location-needs-move`.
+
+- **Keys and marker.** A key is a slug plus a hash of the source's identity. The folder's
+  `.promptmanager-thumbnails.json` stores each key with its identity and install. Only a single `<slug>-<hex>` name
+  is ever joined onto the folder, and anything else in a marker is dropped. Every write and delete passes
+  `preview_store.may_write`: below a key folder, with no link, junction or reparse point on the way
+  (`tests/test_preview_store_folder.py`, including tampered markers).
+- **Unavailable.** A folder is unavailable when it is missing, is a link, holds no marker or holds another
+  folder's marker. A probe checks it every 5 s, and a store call that fails or stalls marks it at once. While it is
+  unavailable:
+  - originals are served and nothing is written anywhere;
+  - generate, convert, clear and prune answer 409 `location-unavailable`;
+  - the folder is never made again except by Recreate folder.
+
+  The configured location is in force before any request. A bad value is `invalid`, never a fall-back to the
+  sources.
+- **Move thumbnails.** The job is a `thumbnails` job with `scope.mode` `move` (`tests/test_thumbnail_move.py`). It
+  runs in this order:
+  1. The review answers `POST /images/thumbnails/plan`; not enough space answers 507.
+  2. A token starts the job.
+  3. Automatic previews are held.
+  4. Copies run on the job pool in chunks of 2 or 4 files, each with its size checked and its mtime rounded up to
+     2 s.
+  5. One config commit makes the switch and records the old stores under `cleanup`. The job cannot be cancelled
+     from here.
+  6. The old copies are removed.
+
+  If it fails or is cancelled before the switch, nothing changes. A crash leaves
+  `<data>/data/thumbnail-move.json`, which the next start undoes. Old copies still held open are removed by the
+  next start's "Move thumbnails (finishing)" job. Leaving a folder also removes the previews of this install's
+  sources that are no longer listed (Q320-8).
+- **Remove thumbnails of removed sources** (`scope.mode` `prune`). It is offered in Thumbnail maintenance only in
+  folder mode, and it touches only this install's keys.
+- **Page tests.** `tests/browser/workspace-thumbnail-location.spec.mjs` covers the row, both moves, the in-place
+  refusal, the confirmation, the notice with Recreate and with Use this folder. The legacy gallery runs once in
+  folder mode (`workflows.spec.mjs`). The hooks are `tests/browser/thumbnail_location_fixture.py`.
+
+Beyond the concept, against the approved static mock `docs/egm/post-3.4.1/320-mock`:
+
+- D196: the Thumbnail location row in System › Sources & scans, after Thumbnail generation.
+  - The status line is "Now: inside each source." or "Now: one folder, <path>.". It does not show the mock's
+    "(3 sources, 8,512 thumbnails)" counts, because counting means walking every preview on each view. It says
+    "Using the default folder next to your library." and names old copies that wait to be removed.
+- D197: the Move thumbnails review uses the shared review.
+  - A grid value may be a lead sentence with a list (`[lead, items]`).
+  - Sources no longer listed appear under "Removed after the move", where the mock has a "Not moved" row
+    (maintainer answer to Q320-8: remove).
+  - The Free space line is "<size> free", without naming the disk.
+- D198: Activity's view of `mode: move` and `mode: prune`.
+  - Tiles for a move: Moved, Already there, Left in place, Errors. Tiles for a prune: Removed, Left in place,
+    Errors.
+  - The step line under the bar is `#job-phase`.
+  - Cancel shows only until the switch.
+- D199: the unavailable notice on Overview and Sources & scans.
+  - Recreate folder appears for a missing folder, after a confirmation.
+  - Use this folder appears for another folder's marker.
+  - Change location focuses the row; on Overview it opens Sources & scans.
+- D200: Thumbnail maintenance's "Remove thumbnails of removed sources" scope, with the plan line and a grid of
+  key folders.
+- D201: Move library's review adds "Thumbnails: Stay in <path>. Move them in System › Sources & scans." in folder
+  mode.
+
 ### Backup and restore contract (#249 System S8)
 
 A restore is staged, then published (`database/restore_stage.py`). The upload is

__init__.py

@@ -102,6 +102,7 @@ try:
 
     def _preview_startup():
         api.start_preview_location()
+        api.recover_thumbnail_move()  # an interrupted move, before the sweep
         api.sweep_thumbnail_temps()
 
     threading.Thread(
@@ -118,6 +119,8 @@ try:
         # Build each output root's Images listing once the server's loop runs,
         # so the first open does not wait for the walk (#348).
         host_app.on_startup.append(api.warm_gallery_listings)
+        # #320: old thumbnails a move left to remove, once it is checked.
+        host_app.on_startup.append(api.finish_thumbnail_move)
     if host_app is not None and hasattr(host_app, "on_shutdown"):
         # Open Needs review change streams end before the server waits on them.
         host_app.on_shutdown.append(api.close_review_events)

py/api/__init__.py

@@ -62,6 +62,10 @@ from .job_routes import JobRoutesMixin
 from .image_tag_jobs import ImageTagJobsMixin
 from .thumbnail_jobs import ThumbnailJobsMixin
 from .thumbnail_location import ThumbnailLocationMixin
+from .thumbnail_move import ThumbnailMoveMixin
+from .thumbnail_move_plan import ThumbnailMovePlanMixin
+from .thumbnail_move_recovery import ThumbnailMoveRecoveryMixin
+from .thumbnail_prune import ThumbnailPruneMixin
 from .thumbnail_convert import ThumbnailConvertMixin
 from .thumbnail_auto import AutoThumbnailMixin
 from .system import SystemRoutesMixin
@@ -341,6 +345,10 @@ class PromptManagerAPI(
     ImageTagJobsMixin,
     ThumbnailJobsMixin,
     ThumbnailLocationMixin,
+    ThumbnailMovePlanMixin,
+    ThumbnailMoveMixin,
+    ThumbnailMoveRecoveryMixin,
+    ThumbnailPruneMixin,
     ThumbnailConvertMixin,
     AutoThumbnailMixin,
     SystemRoutesMixin,
@@ -719,6 +727,7 @@ class PromptManagerAPI(
         # ── Register domain-specific routes from mixins ───────────────
 
         self._register_pair_routes(routes)
+        self._register_thumbnail_move_routes(routes)
         self._register_library_routes(routes)
         self._register_library_delete_routes(routes)
         self._register_vocabulary_routes(routes)

py/api/library_location.py

@@ -75,6 +75,14 @@ def _failure(message, status, code):
     )
 
 
+def _thumbnail_folder():
+    """The thumbnail folder in use, or None (previews inside each source)."""
+    from . import preview_store
+
+    setting = preview_store.setting()
+    return setting.path if setting.mode == preview_store.FOLDER_MODE else None
+
+
 def _inside_thumbnail_folder(path):
     """Whether *path* lies in the thumbnail folder in use (#320)."""
     from . import preview_store
@@ -236,6 +244,8 @@ class LibraryLocationRoutesMixin:
             ),
             "problem": location.problem or self._node_folder_risk(location),
             "size_bytes": size,
+            # #320 D201: thumbnails kept in one folder stay there (D8A).
+            "thumbnails_folder": _thumbnail_folder(),
         }
 
     # The node folder (tests point it at a temp tree).

py/api/preview_location.py

@@ -29,8 +29,10 @@ import posixpath
 import re
 import stat as stat_module
 import threading
+import time
 import unicodedata
 import uuid
+from contextlib import contextmanager
 from dataclasses import dataclass, field
 
 try:
@@ -44,6 +46,11 @@ SOURCE_MODE = "source"
 FOLDER_MODE = "folder"
 MARKER_NAME = ".promptmanager-thumbnails.json"
 MARKER_VERSION = 1
+# Beside the marker while one process re-reads, changes and replaces it.
+MARKER_LOCK_NAME = ".promptmanager-thumbnails.lock"
+MARKER_LOCK_TIMEOUT = 1.5  # a writer waits this long (under preview_store.IO_BUDGET)
+MARKER_LOCK_STALE = 30.0  # a lock this old was left by a writer that died
+MARKER_LOCK_POLL = 0.02
 # 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.
@@ -96,6 +103,11 @@ MESSAGES = {
         "This folder contains an image source. Choose a folder beside it instead.",
     ),
     "location-same": (400, "Thumbnails are already stored here."),
+    "location-inside-current": (
+        400,
+        "This folder is inside the thumbnail folder in use. Choose a folder "
+        "outside it.",
+    ),
     "location-parent-missing": (
         400,
         "The folder that would hold it doesn't exist. Check the path and that its "
@@ -119,6 +131,11 @@ MESSAGES = {
         400,
         "This path is too long for Windows. Choose a shorter folder path.",
     ),
+    "insufficient-space": (
+        507,
+        "There is not enough space for the thumbnails. Free some space or choose "
+        "another folder.",
+    ),
     "location-needs-move": (
         409,
         "Use Move thumbnails in System › Sources & scans.",
@@ -408,10 +425,70 @@ def valid_keys(keys):
     return kept
 
 
+@contextmanager
+def _marker_lock(folder):
+    """Hold the folder's marker lock: a file made with ``O_CREAT | O_EXCL``
+    (atomic on local disks and SMB/NFS shares, unlike ``flock``), so installs
+    sharing the folder change the marker one at a time. A lock older than
+    MARKER_LOCK_STALE was left by a writer that died and is taken over;
+    waiting longer than MARKER_LOCK_TIMEOUT raises TimeoutError."""
+    lock = os.path.join(folder, MARKER_LOCK_NAME)
+    deadline = time.monotonic() + MARKER_LOCK_TIMEOUT
+    while True:
+        try:
+            handle = os.open(lock, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o644)
+        except FileExistsError:
+            _break_stale_lock(lock)
+            if time.monotonic() >= deadline:
+                raise TimeoutError(
+                    errno.ETIMEDOUT, "the thumbnail folder marker is locked", lock
+                )
+            time.sleep(MARKER_LOCK_POLL)
+            continue
+        try:
+            os.write(handle, f"{os.getpid()}\n".encode("ascii"))
+        finally:
+            os.close(handle)  # closed before any unlink (Windows)
+        break
+    try:
+        yield
+    finally:
+        try:
+            os.unlink(lock)
+        except FileNotFoundError:
+            pass  # taken over as stale by another writer: nothing to release
+
+
+def _break_stale_lock(lock):
+    """Remove *lock* when it is older than MARKER_LOCK_STALE (its writer
+    died); a lock that changed since it was looked at is left alone."""
+    try:
+        before = os.stat(lock)
+    except FileNotFoundError:
+        return
+    if time.time() - before.st_mtime < MARKER_LOCK_STALE:
+        return
+    logger.warning(f"Taking over a stale thumbnail folder lock: {lock}")
+    try:
+        after = os.stat(lock)
+        if (after.st_ino, after.st_mtime_ns) == (before.st_ino, before.st_mtime_ns):
+            os.unlink(lock)
+    except FileNotFoundError:
+        pass  # another writer took it over first
+
+
 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."""
+    (a temp sibling renamed over it), all under the folder's marker lock so a
+    second install sharing the folder never drops this one's change (and
+    the other way round). Callers hold ``preview_store.write_lock`` (threads
+    of this process). Returns the marker written; TimeoutError when another
+    writer holds the lock too long."""
+    with _marker_lock(folder):
+        return _write_marker_locked(folder, mutate)
+
+
+def _write_marker_locked(folder, mutate):
     current = read_marker(folder) or {}
     marker = {
         "version": MARKER_VERSION,
@@ -526,7 +603,7 @@ def _timed(func, timeout):
     def run():
         try:
             result["value"] = func()
-        except BaseException as error:  # handed to the caller
+        except BaseException as error:  # _timed() re-raises it in the caller
             result["error"] = error
         finally:
             done.set()
@@ -620,6 +697,14 @@ def _filesystem_problem(path, purpose, context, default):
         )
     ):
         return "location-same"
+    if (
+        purpose == "move"
+        and context.current_path
+        and _inside(path, [context.current_path])
+    ):
+        # The old key folders are removed after the move, and with them a
+        # target nested inside one of them.
+        return "location-inside-current"
     # 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):

py/api/preview_store.py

@@ -186,6 +186,13 @@ def _set_keys(parsed, marker):
             _key_cache.clear()
 
 
+def set_cleanup(entries):
+    """The setting's ``cleanup`` after a removal was committed."""
+    global _parsed
+    with _state_lock:
+        _parsed = replace(_parsed, cleanup=tuple(entries))
+
+
 def mark_invalid():
     """The setting cannot be used (a startup check that failed): folder
     mode stays in force, unavailable, so nothing is written anywhere."""
@@ -256,22 +263,40 @@ class _Probe:
         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."""
+        """``folder_state`` within IO_BUDGET; raises TimeoutError.
+
+        The answer comes from a check that started after this call: one
+        already running may have looked before what the caller just did, so
+        it is waited for (never run twice at once; a stuck one times out here)
+        and followed by a new one."""
+        key = (path, expected_id)
         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")
+            earlier = self._running(key)
+            pending = None if earlier else self._start_check(path, expected_id)
+        if earlier:
+            self._wait(earlier)
+            with self._lock:
+                # One another caller started meanwhile is new enough to join.
+                pending = self._running(key) or self._start_check(path, expected_id)
+        self._wait(pending)
         with self._lock:
             if self._checking is pending:
                 self._checking = None
-        if "error" in result:
-            raise result["error"]
-        return result["value"]
+        if "error" in pending[2]:
+            raise pending[2]["error"]
+        return pending[2]["value"]
+
+    def _running(self, key):
+        """The check of *key* still running (lock held), else None."""
+        pending = self._checking
+        if pending is None or pending[0] != key or pending[1].is_set():
+            return None
+        return pending
+
+    @staticmethod
+    def _wait(pending):
+        if not pending[1].wait(IO_BUDGET):
+            raise TimeoutError("the thumbnail folder did not answer in time")
 
     def _start_check(self, path, expected_id):
         """Start one check on its own thread (lock held)."""
@@ -281,7 +306,7 @@ class _Probe:
         def run():
             try:
                 result["value"] = preview_location.folder_state(path, expected_id)
-            except BaseException as error:  # handed to the caller
+            except BaseException as error:  # check() re-raises it in the caller
                 result["error"] = error
             finally:
                 done.set()
@@ -407,7 +432,9 @@ def _key_folder(folder, key):
 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."""
+    the thumbnail folder itself). None while the folder is unavailable, and
+    for a key folder that is a link, junction or reparse point: nothing is
+    planned, written or removed through it."""
     location = location or current()
     store = store_dir(root, location)
     if location.mode == SOURCE_MODE or (store is None and not location.available):
@@ -416,12 +443,20 @@ def ensure_store(root, location=None, *, create=False):
         store = _assign_key(root, location)
         if store is None:
             return None
+    try:
+        with store_io():
+            linked = preview_location.is_link_like(store)
+    except OSError:
+        return None
+    if linked:
+        logger.warning(f"Skipping the thumbnail folder {store}: it is a link")
+        return None
     if create:
         try:
             with store_io():
                 os.mkdir(store)
         except FileExistsError:
-            pass
+            pass  # already there: another writer made the key folder
         except OSError:
             return None
     return store
@@ -521,7 +556,7 @@ def make_parent(dst, location=None):
             try:
                 os.mkdir(current_folder)
             except FileExistsError:
-                pass
+                pass  # already there: another writer made this folder
             if preview_location.is_link_like(current_folder):
                 raise PermissionError(errno.EACCES, "a link in the thumbnail folder")
 

py/api/settings_routes.py

@@ -491,7 +491,8 @@ class SettingsRoutesMixin:
             for key in _THUMBNAIL_SETTINGS
         )
         job = self._jobs.active(THUMBNAILS_JOB_KIND) if changed else None
-        if job is None or job.scope.get("mode") != "convert":
+        # A move copies previews of the current settings (#320).
+        if job is None or job.scope.get("mode") not in ("convert", "move"):
             return None
         return web.json_response(
             {

py/api/thumbnail_auto.py

@@ -775,6 +775,9 @@ class AutoThumbnailMixin:
                 return job
             if job.kind == THUMBNAILS_JOB_KIND and job.title == THUMBNAIL_CLEAR_TITLE:
                 return job
+            # Move thumbnails holds automatic previews for its whole run (D2A).
+            if job.kind == THUMBNAILS_JOB_KIND and job.scope.get("mode") == "move":
+                return job
         return None
 
     # ── the view admission ────────────────────────────────────────────

py/api/thumbnail_convert.py

@@ -44,7 +44,7 @@ import os
 import time
 from pathlib import Path
 
-from . import thumbnail_settings
+from . import preview_store, thumbnail_settings
 from .thumbnail_encode import ThumbnailVerifyError, verify_thumbnail, writer_settings
 from .admin import _plural, _scan_location, _scan_roots, _scan_status_line
 from .images import _thumbnail_is_fresh, _worker_threads, _thumbnail_batch_size
@@ -393,6 +393,12 @@ class ThumbnailConvertMixin:
         removed, with no bytes (another run or a sibling removed it)."""
         removed, kept = 0, []
         for path, size in old:
+            if not preview_store.may_write(path):
+                # Folder mode: never a file outside a key folder, or one
+                # reached through a link (checked at the moment of the delete).
+                self.logger.warning(f"Not removing {path}: outside the thumbnails")
+                kept.append(path)
+                continue
             try:
                 os.remove(path)
             except FileNotFoundError:

py/api/thumbnail_encode.py

@@ -46,7 +46,7 @@ from pathlib import Path
 
 from PIL import Image
 
-from . import preview_store, thumbnail_settings
+from . import preview_location, preview_store, thumbnail_settings
 
 try:
     from ...utils.logging_config import get_logger
@@ -114,6 +114,7 @@ def replace_unless_held(tmp, dst, guard=None):
 
 # ``temp_sibling``'s names: ``.<stem>.<tag>-<pid>-<thread id><ext>``.
 _TEMP_NAME = re.compile(r"^\..+\.(?:part|src)-(\d+)-\d+\.[A-Za-z0-9]+$")
+_MOVE_TEMP_NAME = re.compile(r"\.pm-move-[0-9a-f]{8}\.tmp$")
 # A temp at least this old (seconds) outlived its write: no encode runs longer.
 STRAY_TEMP_AGE = 4 * ENCODE_TIMEOUT
 # When this process started (this module's import): a temp carrying its PID
@@ -125,7 +126,10 @@ def sweep_stray_temps(folder, min_age=None):
     """Remove the temp files a crashed write left under *folder* (a preview
     folder, walked without following links); returns ``(count, bytes)``.
 
-    Only ``temp_sibling`` names are touched, never a link. With *min_age*
+    Only ``temp_sibling`` names are touched, never a link: a folder that is a
+    link, junction or reparse point is not walked into (Windows walks into a
+    junction even without ``followlinks``), and each file passes
+    ``preview_store.may_write`` before it goes. With *min_age*
     (seconds, the startup sweep) a temp younger than that, or one this
     process wrote since it started (it may still be writing it), is left;
     ``min_age=0`` (the clear, which holds every preview write) removes them
@@ -135,19 +139,39 @@ def sweep_stray_temps(folder, min_age=None):
     now = time.time()
     own = str(os.getpid())
     count = size = 0
-    for root, _dirs, files in os.walk(folder):
+    try:
+        if preview_location.is_link_like(folder):
+            return count, size
+    except OSError as exc:
+        logger.debug(f"Not sweeping {folder}: {exc}")
+        return count, size
+    for root, dirs, files in os.walk(folder):
+        dirs[:] = [d for d in dirs if not _link_like(Path(root) / d)]
         for name in files:
             match = _TEMP_NAME.match(name)
-            if match is None:
+            # A move's copy temp (#320): ``<name>.pm-move-<8 hex>.tmp``.
+            move_temp = match is None and _MOVE_TEMP_NAME.search(name)
+            if match is None and not move_temp:
                 continue
             path = Path(root) / name
             try:
                 stat = path.lstat()
-                if path.is_symlink() or not path.is_file():
+                if (
+                    path.is_symlink()
+                    or not path.is_file()
+                    or _link_like(path)
+                    or not preview_store.may_write(path)
+                ):
                     continue
                 if min_age and (
                     now - stat.st_mtime < min_age
-                    or (match.group(1) == own and stat.st_mtime >= PROCESS_STARTED)
+                    or move_temp
+                    and stat.st_mtime >= PROCESS_STARTED
+                    or (
+                        match is not None
+                        and match.group(1) == own
+                        and stat.st_mtime >= PROCESS_STARTED
+                    )
                 ):
                     continue
                 path.unlink()
@@ -159,6 +183,14 @@ def sweep_stray_temps(folder, min_age=None):
     return count, size
 
 
+def _link_like(path):
+    """``is_link_like``, with a path that cannot be checked counted as one."""
+    try:
+        return preview_location.is_link_like(path)
+    except OSError:
+        return True
+
+
 def _unlink_quietly(path):
     try:
         Path(path).unlink()

py/api/thumbnail_jobs.py

@@ -132,7 +132,10 @@ class ThumbnailJobsMixin:
     """The thumbnail plan, the thumbnail job and the legacy stream on it."""
 
     async def get_thumbnail_plan(self, request):
-        """What a thumbnail run would do (read-only; capped like a run)."""
+        """What a thumbnail run would do (read-only; capped like a run).
+        ``mode=prune`` lists the removed sources' thumbnails (#320)."""
+        if request.query.get("mode") == "prune":
+            return await self.get_prune_plan()
         options, error = thumbnail_options(request.query, from_query=True)
         if error:
             return _invalid(error)
@@ -180,6 +183,8 @@ class ThumbnailJobsMixin:
                 )
             except (ValueError, UnicodeDecodeError):
                 return _invalid(_OPTIONS_ERROR)
+        if isinstance(body, dict) and body.get("mode") in ("move", "prune"):
+            return await self._start_location_job(body)
         options, error = thumbnail_options(body)
         if error:
             return _invalid(error)
@@ -192,6 +197,25 @@ class ThumbnailJobsMixin:
             return busy_response(busy)
         return web.json_response({"success": True, "job": job.summary()}, status=202)
 
+    async def _start_location_job(self, body):
+        """Move thumbnails or Remove thumbnails of removed sources (#320)."""
+        if body["mode"] == "prune":
+            refused = location_refusal()
+            if refused is not None:
+                return refused
+        try:
+            if body["mode"] == "move":
+                started = self.start_thumbnail_move(body)
+            else:
+                started = self.start_thumbnail_prune()
+        except JobBusy as busy:
+            return busy_response(busy)
+        if isinstance(started, web.Response):
+            return started
+        return web.json_response(
+            {"success": True, "job": started.summary()}, status=202
+        )
+
     def _start_thumbnail_job(self, options):
         """Register the thumbnail job synchronously; raises JobBusy.
 
@@ -515,8 +539,12 @@ class ThumbnailJobsMixin:
         the way.
         """
         running = self._jobs.active(THUMBNAILS_JOB_KIND)
-        if running is not None and running.scope.get("mode") == "convert":
-            return None, False, running  # no legacy frames to follow (#317)
+        if running is not None and running.scope.get("mode") in (
+            "convert",
+            "move",
+            "prune",
+        ):
+            return None, False, running  # no legacy frames to follow (#317, #320)
         if running is not None and running.cancellable:
             return running, False, None
         busy = self._jobs.blocker(THUMBNAILS_JOB_KIND, admission="guarded")

py/api/thumbnail_location.py

@@ -82,7 +82,7 @@ class ThumbnailLocationMixin:
                 if folder:
                     contained.append(str(folder))
         except ImportError:
-            pass
+            pass  # folder_paths exists only inside ComfyUI (tests run without it)
         try:
             contained.append(library_location.config_location()[1])
         except Exception:  # no data folder can be decided
@@ -157,4 +157,6 @@ class ThumbnailLocationMixin:
         if with_path:
             payload["path"] = location.path
             payload["default_path"] = default_path()
+            # Old copies a move still has to remove (#320 Part 3).
+            payload["cleanup_pending"] = len(preview_store.setting().cleanup)
         return payload

py/api/thumbnail_move.py

@@ -0,0 +1,579 @@
+"""Move thumbnails (#320 Part 3): the reviewed job that changes where previews
+live, the removal of old copies after it, recovery at startup and the
+"Remove thumbnails of removed sources" scope (Part 5).
+
+Kind ``thumbnails`` with ``mode: "move"`` (D1A): admission ``guarded``, so
+scans, deletes, tagging, source changes and every other thumbnail job wait
+for it, while the monitor's Needs review recording runs beside it.
+Automatic previews are held for the whole job (D2A, ``_auto_thumbnail_blocker``).
+
+Phases: review (``POST /images/thumbnails/plan``, no writes but the
+validation probe), start (a plan token), hold (generation bump, journal,
+target folder), copy (cancellable at every file boundary, on the job pool
+in small chunks, D3A), switch (one config commit sets the location and
+records the old stores under ``cleanup``; not cancellable from here),
+remove old copies, end. Per-file errors switch anyway (D10A); a folder-wide
+error or a cancel before the switch removes what the job wrote and changes
+nothing. The move is the only writer of the setting (D4B).
+
+Moving away from a thumbnail folder also removes the previews of this
+install's sources that are no longer listed (Q320-8: remove).
+"""
+
+import functools
+import os
+import time
+from pathlib import Path
+
+from aiohttp import web
+
+from . import preview_location, preview_store
+from . import thumbnail_move_files as files
+from .jobs import THUMBNAILS_JOB_KIND
+from .thumbnail_move_plan import (
+    MOVE_GROUPS,
+    MOVE_TITLE,
+    PLAN_TTL,
+    _number,
+    _plural,
+    _remove,
+    _roots_revision,
+    _volume,
+    _is_network,
+    _write_json,
+)
+from . import thumbnail_move_plan as move_plan
+
+try:
+    from ...utils import exit_signal
+    from ...utils.parallel import map_parallel
+except ImportError:  # tests import the package top-level
+    from utils import exit_signal
+    from utils.parallel import map_parallel
+
+
+class ThumbnailMoveMixin:
+    """Start and run the move job; the removal of old copies."""
+
+    # ── start ─────────────────────────────────────────────────────────
+
+    def _take_move_plan(self, body):
+        """The plan a start names, or ``(status, code)``."""
+        token = body.get("token")
+        plan = (self._move_plans or {}).get(token) if isinstance(token, str) else None
+        if plan is None or time.monotonic() - plan.created > PLAN_TTL:
+            return 409, "plan-stale"
+        roots = [str(root) for root in self._get_all_output_dirs()]
+        if _roots_revision(roots) != plan.revision:
+            return 409, "plan-stale"
+        if preview_store.setting() != plan.old:
+            return 409, "plan-stale"
+        if plan.confirm_required and body.get("confirmed") is not True:
+            return 400, "confirmation-required"
+        return plan
+
+    def start_thumbnail_move(self, body):
+        """Start the move job (sync, on the loop); raises JobBusy. Returns
+        the job or a refusal response."""
+        if not self._move_recovered:
+            blocker = self._jobs.active(THUMBNAILS_JOB_KIND)
+            return web.json_response(
+                {
+                    "success": False,
+                    "code": "busy",
+                    "error": "Prompt Manager is still checking an earlier move. "
+                    "Try again in a moment.",
+                    "running": None if blocker is None else {"id": blocker.id},
+                },
+                status=409,
+            )
+        taken = self._take_move_plan(body)
+        if isinstance(taken, tuple):
+            status, code = taken
+            text = {
+                "plan-stale": "The review is out of date. Review the move again.",
+                "confirmation-required": "Confirm that Prompt Manager may "
+                "create its folders there.",
+            }[code]
+            return web.json_response(
+                {"success": False, "code": code, "error": text}, status=status
+            )
+        plan = taken
+        self._move_plans.pop(plan.token, None)
+        return self._jobs.start(
+            THUMBNAILS_JOB_KIND,
+            MOVE_TITLE,
+            admission="guarded",
+            cancellable=True,
+            scope={
+                "mode": "move",
+                "from": plan.old.mode,
+                "to": plan.target["mode"],
+                "to_path": plan.target["path"],
+            },
+            groups=MOVE_GROUPS,
+            runner=functools.partial(self._run_thumbnail_move, plan=plan),
+        )
+
+    # ── the job ───────────────────────────────────────────────────────
+
+    async def _run_thumbnail_move(self, job, *, plan):
+        state = {"written": [], "created": [], "created_folder": False}
+        try:
+            await self._run_in_executor(self._move_hold, job, plan, state)
+        except (OSError, ValueError) as error:
+            await self._run_in_executor(self._move_rollback, plan, state)
+            self._end_move(
+                job, "failed", "The new folder could not be made; nothing changed."
+            )
+            self.logger.warning(f"Move thumbnails could not start: {error}")
+            return
+        left = [
+            {
+                "path": "/".join(item),
+                "root": s.label,
+                "detail": "Not a thumbnail; left in place",
+            }
+            for s in plan.sources
+            if s.scan is not None
+            for item in s.scan.left
+        ]
+        if left:
+            job.record_counted("left", len(left), left)
+        items = [
+            (source, item)
+            for source in plan.sources
+            if source.scan is not None
+            for item in source.scan.copy
+        ]
+        total = len(items)
+        job.update(total=total, processed=0, progress=0)
+        chunk = await self._run_in_executor(self._move_chunk, plan)
+        workers = self._move_workers()
+        done = 0
+        for start in range(0, total, chunk):
+            if job.cancelled:
+                await self._move_cancelled(job, plan, state)
+                return
+            batch = items[start : start + chunk]
+            try:
+                results = await self._run_in_executor(
+                    map_parallel,
+                    functools.partial(self._move_copy_one, state=state),
+                    batch,
+                    workers,
+                )
+            except exit_signal.ExitRequested:
+                return  # exiting: the next start undoes the copies (journal)
+            except files.FolderGone as error:
+                self.logger.warning(f"Move thumbnails stopped: {error}")
+                await self._run_in_executor(self._move_rollback, plan, state)
+                self._end_move(
+                    job,
+                    "failed",
+                    "The new folder could not be reached or is full; nothing changed.",
+                )
+                return
+            except Exception:
+                # Anything else before the switch: undo, never half a move.
+                self.logger.exception("Move thumbnails failed while copying")
+                await self._run_in_executor(self._move_rollback, plan, state)
+                self._end_move(job, "failed", "The copy failed; nothing changed.")
+                return
+            for (source, item), (outcome, detail) in zip(batch, results):
+                row = {
+                    "path": "/".join(item[0]),
+                    "root": source.label,
+                    "detail": detail,
+                }
+                job.record(
+                    "moved" if outcome == "moved" else outcome,
+                    row,
+                    f"{outcome}: {source.label} · {'/'.join(item[0])}",
+                )
+            done += len(batch)
+            label = batch[-1][0].label
+            job.update(
+                processed=done,
+                progress=int(done * 90 / total) if total else 90,
+                status_line=(
+                    f"Copying {_number(done)} of {_number(total)} previews · {label}"
+                ),
+            )
+            job.publish({"type": "progress", "summary": job.summary()})
+        if job.cancelled:
+            await self._move_cancelled(job, plan, state)
+            return
+        # The switch: not cancellable from here (an exit still stops between
+        # files and leaves the cleanup to the next start).
+        job.cancellable = False
+        job.update(status_line="Switching to the new location", progress=92)
+        job.publish({"type": "progress", "summary": job.summary()})
+        try:
+            await self._run_in_executor(self._move_switch, plan)
+        except Exception:
+            self.logger.exception("Move thumbnails could not switch")
+            await self._run_in_executor(self._move_rollback, plan, state)
+            self._end_move(
+                job, "failed", "The setting could not be saved; nothing changed."
+            )
+            return
+        await self._remove_old_copies(job, plan)
+        moved = job.counters.get("moved", 0) + job.counters.get("skipped", 0)
+        errors = job.counters.get("errors", 0)
+        where = plan.target["path"] or "inside each source"
+        parts = [f"{_number(moved)} moved"]
+        if errors:
+            parts.append(_plural(errors, "error"))
+        parts.append(
+            "old copies removed"
+            if not job.scope.get("cleanup_left")
+            else "some old copies are removed at the next start"
+        )
+        self._end_move(
+            job, "finished", f"Thumbnails moved to {where}: " + " · ".join(parts) + "."
+        )
+
+    def _end_move(self, job, state, line):
+        job.update(status_line=line, progress=100)
+        job.log_line(line)
+        job.publish({"type": "complete", "summary": job.summary()})
+        job.finish(state)
+
+    def _move_workers(self):
+        from .images import _worker_threads
+
+        return max(1, int(_worker_threads() or 1))
+
+    def _move_chunk(self, plan):
+        """2 files at a time across volumes or to/from a share, else 4, never
+        more than Worker threads (D3A: the pool keeps its size)."""
+        sides = [plan.target["path"] or ""] + [s.old_store or "" for s in plan.sources]
+        sides += [s.new_store for s in plan.sources]
+        network = any(_is_network(side) for side in sides if side)
+        volumes = {_volume(side) for side in sides if side}
+        size = 2 if network or len(volumes) > 1 else 4
+        return max(1, min(size, self._move_workers()))
+
+    def _move_hold(self, job, plan, state):
+        """Phase 3: bump the generation, write the journal, make the target."""
+        preview_store.bump_generation()
+        if plan.to_folder:
+            folder = plan.target["path"]
+            if not os.path.isdir(folder):
+                if plan.target_default:
+                    os.makedirs(folder, exist_ok=True)
+                else:
+                    os.mkdir(folder)
+                state["created_folder"] = True
+            plan.target_id = preview_location.init_folder(
+                folder, created_by_pm=state["created_folder"]
+            )
+            install = plan.install
+
+            def add_keys(marker):
+                for source in plan.sources:
+                    if source.key and source.key not in marker["keys"]:
+                        marker["keys"][source.key] = {
+                            "path": source.root,
+                            "identity": os.path.normcase(os.path.realpath(source.root)),
+                            "dev_ino": preview_location.file_identity(source.root),
+                            "install": install,
+                        }
+
+            with preview_store.write_lock:
+                preview_location.write_marker(folder, add_keys)
+        _write_json(
+            move_plan.journal_path(),
+            {
+                "state": "copying",
+                "from": plan.old.to_config(),
+                "to": plan.target,
+                "stores": [s.new_store for s in plan.sources],
+                # Stores that do not exist yet are made by this job only:
+                # after a crash everything in them goes (B4).
+                "created_dirs": [
+                    s.new_store for s in plan.sources if not os.path.isdir(s.new_store)
+                ],
+                "created_folder": state["created_folder"],
+                "install": plan.install,
+            },
+        )
+        job.log_line(
+            f"Move thumbnails started: {plan.old.path or 'inside each source'} → "
+            f"{plan.target['path'] or 'inside each source'}"
+        )
+
+    def _move_copy_one(self, pair, *, state):
+        """Copy one preview (a pool thread); ``(group, detail)``."""
+        source, (relative, size, mtime_ns) = pair
+        destination = Path(source.new_store).joinpath(*relative)
+        try:
+            if not os.path.isdir(destination.parent):
+                base = Path(source.new_store)
+                if not base.is_dir():
+                    if plan_parent_missing(base):
+                        raise files.FolderGone("the new folder is gone")
+                    created = []
+                    files._make_folders(base.parent, (base.name,), created)
+                    state["created"].extend(created)
+                created = []
+                files._make_folders(base, relative[:-1], created)
+                state["created"].extend(created)
+            # Also when the folders already exist (Codex P1 on #592).
+            files.refuse_links(source.new_store, relative[:-1])
+            outcome = files.copy_one(
+                Path(source.old_store).joinpath(*relative),
+                destination,
+                size,
+                mtime_ns,
+                state["created"],
+            )
+        except files.CopyFailed as error:
+            return "errors", str(error)
+        if outcome == "moved":
+            state["written"].append(str(destination))
+            return "moved", "Copied and checked"
+        return "skipped", "Already there"
+
+    async def _move_cancelled(self, job, plan, state):
+        if exit_signal.requested():
+            return  # the next start removes the partial copies (journal)
+        await self._run_in_executor(self._move_rollback, plan, state)
+        self._end_move(job, "cancelled", "Move cancelled: nothing changed.")
+
+    def _move_rollback(self, plan, state):
+        """Remove what this job wrote (files, folders it made, a folder and
+        marker it created) and the journal: nothing changed."""
+        for path in state["written"]:
+            files.remove_file(path)
+        for folder in reversed(state["created"]):
+            files.remove_empty_folders(folder)
+        if plan.to_folder and state.get("created_folder"):
+            folder = plan.target["path"]
+            try:
+                names = os.listdir(folder)
+            except OSError:
+                names = None
+            if names is not None and set(names) <= {preview_location.MARKER_NAME}:
+                _remove(preview_location.marker_path(folder))
+                files.remove_empty_folders(folder)
+        _remove(move_plan.journal_path())
+
+    def _move_switch(self, plan):
+        """Phase 5: one config commit sets the location and records the old
+        stores under ``cleanup``; the runtime follows under the write lock."""
+        from ..config import PromptManagerConfig
+        from ..config_store import get_config_store
+
+        entry = {
+            "from": plan.old.mode,
+            "folder": plan.old.path if plan.from_folder else None,
+            "stores": [s.old_store for s in plan.sources if s.old_store],
+            "drop_keys": self._old_keys_to_drop(plan),
+            "install": plan.install,
+        }
+        # An older entry never removes a store the move just filled (a move
+        # back while old copies still waited).
+        filled = {os.path.normcase(s.new_store) for s in plan.sources}
+        cleanup = []
+        for old_entry in plan.old.cleanup:
+            stores = [
+                store
+                for store in old_entry.get("stores") or []
+                if isinstance(store, str) and os.path.normcase(store) not in filled
+            ]
+            folder = old_entry.get("folder")
+            drop = [
+                key
+                for key in old_entry.get("drop_keys") or []
+                if not folder
+                or os.path.normcase(os.path.join(folder, key)) not in filled
+            ]
+            cleanup.append({**old_entry, "stores": stores, "drop_keys": drop})
+        cleanup.append(entry)
+        if plan.to_folder:
+            value = {
+                "mode": preview_store.FOLDER_MODE,
+                "path": plan.target["path"],
+                "id": plan.target_id,
+                "install": plan.install,
+                "cleanup": cleanup,
+            }
+        else:
+            value = {
+                "mode": preview_store.SOURCE_MODE,
+                "install": plan.install,
+                "cleanup": cleanup,
+            }
+
+        def mutate(config):
+            section = config.get("thumbnails")
+            if not isinstance(section, dict):
+                section = {}
+            section["location"] = value
+            config["thumbnails"] = section
+            return config
+
+        def publish(_config):
+            PromptManagerConfig.THUMBNAIL_LOCATION = value
+            preview_store.configure(value, check=False)
+
+        get_config_store().commit(mutate, expected_revision=None, publish=publish)
+        # The switch is made: what follows never undoes it.
+        try:
+            preview_store.refresh()
+            _remove(move_plan.journal_path())
+        except OSError as error:
+            self.logger.warning(
+                f"The thumbnail move journal stays until the next start: {error}"
+            )
+
+    def _old_keys_to_drop(self, plan):
+        if not plan.from_folder:
+            return []
+        keys = [s.old_store and os.path.basename(s.old_store) for s in plan.sources]
+        keys = [key for key in keys if key]
+        if plan.to_folder and os.path.normcase(plan.old.path or "") == os.path.normcase(
+            plan.target["path"] or ""
+        ):
+            return []
+        return keys + [key for key, _store, _scan in plan.unlisted]
+
+    async def _remove_old_copies(self, job, plan):
+        """Phase 6: run the cleanup the switch recorded (and earlier ones)."""
+        job.update(status_line="Removing old copies", progress=95)
+        job.publish({"type": "progress", "summary": job.summary()})
+        left = await self._run_in_executor(self._run_cleanup, job)
+        if left:
+            job.scope["cleanup_left"] = True
+
+    def _run_cleanup(self, job=None):
+        """Remove every recorded old store's previews, temps and empty
+        folders; keep in ``cleanup`` what could not go. Returns how many
+        entries are left (blocking)."""
+        setting = preview_store.setting()
+        remaining = []
+        for entry in setting.cleanup:
+            if not self._cleanup_entry(entry, job):
+                remaining.append(entry)
+        if list(setting.cleanup) != remaining:
+            self._commit_cleanup(remaining)
+        return len(remaining)
+
+    def _cleanup_entry(self, entry, job):
+        """One recorded removal; True when it is complete."""
+        folder = entry.get("folder")
+        stores = [
+            store for store in entry.get("stores") or [] if isinstance(store, str)
+        ]
+        if folder:
+            if not os.path.isdir(folder):
+                return False  # the old folder is not reachable: kept for later
+            folder_marker = preview_location.read_marker(folder) or {}
+            keys = folder_marker.get("keys") or {}
+            stores += [
+                os.path.join(folder, key)
+                for key in entry.get("drop_keys") or []
+                if preview_location.valid_key(key) and key in keys
+            ]
+        complete = True
+        for store in dict.fromkeys(stores):
+            if not self._remove_store(store, folder, job):
+                complete = False
+        if folder and complete:
+            drop = set(entry.get("drop_keys") or [])
+
+            def forget(marker):
+                for key in drop:
+                    marker["keys"].pop(key, None)
+
+            try:
+                with preview_store.write_lock:
+                    marker = preview_location.write_marker(folder, forget)
+            except (OSError, ValueError):
+                return False
+            if not marker["keys"] and marker.get("created_by_pm"):
+                names = set(os.listdir(folder))
+                rest = {n for n in names if not files.is_os_metadata(n)}
+                if rest <= {preview_location.MARKER_NAME}:
+                    _remove(preview_location.marker_path(folder))
+                    files.remove_empty_folders(folder)
+        return complete
+
+    def _remove_store(self, store, folder, job):
+        """Remove the previews and temps of one old store; True when done."""
+        store = Path(store)
+        if folder:
+            # A key folder directly in the old thumbnail folder, nothing else.
+            if not preview_location.valid_key(store.name) or os.path.normcase(
+                str(store.parent)
+            ) != os.path.normcase(str(folder)):
+                return True
+        elif store.name != "thumbnails":
+            return True  # a hand-edited entry never names another folder
+        if not store.exists():
+            return True
+        if preview_location.is_link_like(store):
+            return True  # never through a link; nothing of ours there
+        complete = True
+        removed = 0
+        for current, dirs, names in os.walk(store):
+            dirs[:] = [
+                d for d in dirs if not preview_location.is_link_like(Path(current) / d)
+            ]
+            for name in names:
+                path = Path(current) / name
+                if files.is_temp_name(name) or files._preview_parts(name):
+                    if preview_location.is_link_like(path):
+                        continue
+                    if exit_signal.requested():
+                        return False  # exiting: the next start finishes it
+                    if files.remove_file(path):
+                        removed += 1
+                    else:
+                        complete = False
+                        relative = path.relative_to(store).as_posix()
+                        self.logger.warning(
+                            f"Could not remove the old thumbnail {path}; it is "
+                            "removed at the next start"
+                        )
+                        if job is not None:
+                            # D10A: a removal that failed is an error the job lists.
+                            job.record(
+                                "errors",
+                                {
+                                    "path": relative,
+                                    "root": store.name,
+                                    "detail": "In use by another program; removed "
+                                    "at the next start",
+                                },
+                                f"In use by another program; removed at the next "
+                                f"start: {store.name}/{relative}",
+                            )
+        files.remove_empty_folders(store)
+        if job is not None and removed:
+            job.log_line(f"Removed {_number(removed)} old files from {store.name}")
+        return complete
+
+    def _commit_cleanup(self, remaining):
+        from ..config import PromptManagerConfig
+        from ..config_store import get_config_store
+
+        def mutate(config):
+            section = config.get("thumbnails")
+            if isinstance(section, dict) and isinstance(section.get("location"), dict):
+                section["location"]["cleanup"] = remaining
+            return config
+
+        def publish(config):
+            value = (config.get("thumbnails") or {}).get("location")
+            PromptManagerConfig.THUMBNAIL_LOCATION = value
+            preview_store.set_cleanup(remaining)
+
+        get_config_store().commit(mutate, expected_revision=None, publish=publish)
+
+
+def plan_parent_missing(store):
+    """A key folder's parent (the thumbnail folder) is gone mid-copy."""
+    return not Path(store).parent.is_dir()

py/api/thumbnail_move_files.py

@@ -0,0 +1,348 @@
+"""The file side of Move thumbnails (#320 Part 3): what is copied, how a
+copy is made and checked, and how old copies are removed.
+
+No job, HTTP or config code lives here; ``thumbnail_move`` drives it.
+
+Copy set (B5): each old store is walked with ``os.scandir``, never through a
+symlink, junction or reparse point (counted as "skipped link"). A regular
+file is
+
+- copied when it has the preview grammar (``<name>_thumb<ext>``, any written
+  or legacy extension) and its original still exists in the source;
+- removed with the old copies, not copied, when it is an orphan (no
+  original), lies under a trash folder, or is a temp sibling of a write
+  (``.<stem>.part-…``) or of a move (``….pm-move-<8 hex>.tmp``);
+- left in place otherwise ("Left in place"). OS metadata files (``Thumbs.db``,
+  ``desktop.ini``, ``.DS_Store``, ``._*``) are removed only when they alone
+  keep a preview folder from being removed.
+
+A copy goes to ``<name>.pm-move-<8 hex>.tmp`` beside its destination, its
+size is compared, its mtime is set to the source preview's rounded up to the
+next 2-second step (FAT/exFAT keep it fresh, a stale preview stays stale),
+then it is renamed into place. Every handle is closed before a rename or an
+unlink (Windows).
+"""
+
+import errno
+import logging
+import os
+import re
+import secrets
+import shutil
+import stat as stat_module
+import time
+from dataclasses import dataclass, field
+from pathlib import Path
+
+from . import preview_location, thumbnail_settings
+from .thumbnail_encode import _TEMP_NAME
+
+try:
+    from ...utils.media_folders import in_trash_folder
+except ImportError:  # tests import the package top-level
+    from utils.media_folders import in_trash_folder
+
+logger = logging.getLogger("prompt_manager.thumbnails")
+
+MOVE_TEMP = re.compile(r"\.pm-move-[0-9a-f]{8}\.tmp$")
+MOVE_TEMP_SUFFIX = ".pm-move-{token}.tmp"
+# The temp name adds this much to a destination path (".pm-move-xxxxxxxx.tmp").
+TEMP_EXTRA = 21
+WINDOWS_FILE_LIMIT = 259
+MTIME_STEP_NS = 2_000_000_000
+# A file another program holds open (Windows) is retried this often.
+REMOVE_RETRIES = (0.5, 1.0, 2.0)
+_WINDOWS_IN_USE = {32, 33}  # ERROR_SHARING_VIOLATION, ERROR_LOCK_VIOLATION
+
+
+def is_os_metadata(name):
+    folded = name.lower()
+    return folded in preview_location.OS_METADATA or folded.startswith("._")
+
+
+def is_temp_name(name):
+    return bool(_TEMP_NAME.match(name) or MOVE_TEMP.search(name))
+
+
+def _preview_parts(name):
+    """``(base, extension)`` of a preview file name, or None."""
+    stem, extension = os.path.splitext(name)
+    suffix = thumbnail_settings.THUMB_SUFFIX
+    if not stem.endswith(suffix) or len(stem) <= len(suffix):
+        return None
+    if extension.lower() not in thumbnail_settings.THUMBNAIL_EXTENSIONS:
+        return None
+    return stem[: -len(suffix)], extension
+
+
+def _has_original(root, rel_dir, base, extension):
+    """Whether a media file in ``<root>/<rel_dir>`` owns this preview."""
+    folder = Path(root).joinpath(*rel_dir)
+    try:
+        with os.scandir(folder) as entries:
+            for entry in entries:
+                stem, source_ext = os.path.splitext(entry.name)
+                if stem != base or not source_ext:
+                    continue
+                if not thumbnail_settings.owns(extension, source_ext):
+                    continue
+                if entry.is_file(follow_symlinks=False):
+                    return True
+    except OSError:
+        return False
+    return False
+
+
+@dataclass
+class StoreScan:
+    """One old store's files, sorted into what the move does with them."""
+
+    store: Path
+    copy: list = field(default_factory=list)  # (relative parts, size, mtime_ns)
+    orphans: list = field(default_factory=list)  # relative parts
+    temps: list = field(default_factory=list)
+    left: list = field(default_factory=list)
+    links: int = 0
+    bytes: int = 0
+
+    @property
+    def removable(self):
+        return [item[0] for item in self.copy] + self.orphans + self.temps
+
+
+def scan_store(store, root, *, stop=None):
+    """Sort the files of *store* (the previews of the source *root*).
+
+    A store that does not exist, or is itself a link, scans empty (a link
+    counts once).
+    """
+    store = Path(store)
+    scan = StoreScan(store)
+    try:
+        if preview_location.is_link_like(store):
+            scan.links += 1
+            return scan
+        if not store.is_dir():
+            return scan
+    except OSError:
+        return scan
+    _walk(scan, store, (), root, stop)
+    return scan
+
+
+def _walk(scan, folder, rel_dir, root, stop):
+    try:
+        with os.scandir(folder) as iterator:
+            entries = list(iterator)
+    except OSError:
+        return
+    for entry in entries:
+        if stop is not None and stop():
+            return
+        relative = rel_dir + (entry.name,)
+        path = Path(entry.path)
+        try:
+            if preview_location.is_link_like(path):
+                scan.links += 1
+                continue
+            info = entry.stat(follow_symlinks=False)
+        except OSError:
+            continue
+        if stat_module.S_ISDIR(info.st_mode):
+            _walk(scan, path, relative, root, stop)
+            continue
+        if not stat_module.S_ISREG(info.st_mode):
+            scan.left.append(relative)
+            continue
+        if is_temp_name(entry.name):
+            scan.temps.append(relative)
+            continue
+        if is_os_metadata(entry.name):
+            continue  # removed only when it alone keeps a folder (rmdir)
+        parts = _preview_parts(entry.name)
+        if parts is None:
+            scan.left.append(relative)
+            continue
+        if in_trash_folder("/".join(relative)) or not _has_original(
+            root, rel_dir, *parts
+        ):
+            scan.orphans.append(relative)
+            continue
+        scan.copy.append((relative, info.st_size, info.st_mtime_ns))
+        scan.bytes += info.st_size
+
+
+# ── copying ──────────────────────────────────────────────────────────
+
+
+def rounded_mtime_ns(mtime_ns):
+    """*mtime_ns* rounded up to the next 2-second step (FAT/exFAT)."""
+    return -(-mtime_ns // MTIME_STEP_NS) * MTIME_STEP_NS
+
+
+def too_long(path, long_paths=None):
+    """A destination (with its temp name) over Windows' file limit."""
+    if os.name != "nt" and long_paths is not False:
+        return False
+    if long_paths is None:
+        long_paths = preview_location.long_paths_enabled()
+    if long_paths:
+        return False
+    return len(str(path)) + TEMP_EXTRA > WINDOWS_FILE_LIMIT
+
+
+class CopyFailed(Exception):
+    """One preview could not be copied (a per-file error, D10A)."""
+
+
+class FolderGone(Exception):
+    """The destination cannot take writes any more (abort the move)."""
+
+
+def _make_folders(base, relative_dir, created):
+    """Make ``base/relative_dir`` one folder at a time (*base* must exist);
+    every folder made is added to *created*."""
+    current = Path(base)
+    for part in relative_dir:
+        current = current / part
+        try:
+            os.mkdir(current)
+            created.append(str(current))
+        except FileExistsError:
+            pass  # already there: made by an earlier copy into this folder
+        except OSError as error:
+            if error.errno in (errno.ENOSPC, errno.EDQUOT, errno.EROFS):
+                raise FolderGone(str(error)) from error
+            raise CopyFailed(
+                "Its folder could not be made; made again on view"
+            ) from error
+        if preview_location.is_link_like(current):
+            raise CopyFailed("a link in the destination")
+
+
+def refuse_links(base, relative_dir):
+    """Raise CopyFailed when *base* or an existing folder on the way to
+    ``base/relative_dir`` is a link (never copy through one)."""
+    current = Path(base)
+    for part in (None, *relative_dir):
+        if part is not None:
+            current = current / part
+        if preview_location.is_link_like(current):
+            raise CopyFailed("a link in the destination")
+
+
+def copy_one(source, destination, size, mtime_ns, created, *, long_paths=None):
+    """Copy one preview; returns ``"moved"`` or ``"skipped"`` (already
+    there). Raises CopyFailed (this file) or FolderGone (the destination)."""
+    destination = Path(destination)
+    if too_long(destination, long_paths):
+        raise CopyFailed("Path too long for Windows; the original is shown")
+    wanted = rounded_mtime_ns(mtime_ns)
+    try:
+        existing = os.stat(destination, follow_symlinks=False)
+    except FileNotFoundError:
+        existing = None
+    except OSError as error:
+        raise FolderGone(str(error)) from error
+    if (
+        existing is not None
+        and stat_module.S_ISREG(existing.st_mode)
+        and existing.st_size == size
+        and existing.st_mtime_ns == wanted
+    ):
+        return "skipped"
+    temp = destination.with_name(
+        destination.name + MOVE_TEMP_SUFFIX.format(token=secrets.token_hex(4))
+    )
+    try:
+        shutil.copyfile(source, temp)
+        if os.stat(temp).st_size != size:
+            raise CopyFailed("The copy did not match; made again on view")
+        os.utime(temp, ns=(wanted, wanted))
+        os.replace(temp, destination)
+    except CopyFailed:
+        _unlink(temp)
+        raise
+    except FileNotFoundError as error:
+        _unlink(temp)
+        if not Path(source).exists():
+            raise CopyFailed("Could not be read; made again on view") from error
+        raise FolderGone(str(error)) from error
+    except OSError as error:
+        _unlink(temp)
+        if error.errno in (errno.ENOSPC, errno.EDQUOT, errno.EROFS):
+            raise FolderGone(str(error)) from error
+        raise CopyFailed("Could not be copied; made again on view") from error
+    return "moved"
+
+
+def _unlink(path):
+    """Remove a copy's temp file after a failed copy; True when it is gone.
+
+    One that cannot go (a Windows lock, permissions) is logged with its path;
+    the next start's temp sweep removes it (``.pm-move-*.tmp``)."""
+    try:
+        os.unlink(path)
+        return True
+    except FileNotFoundError:
+        return True  # never written, or already renamed into place
+    except OSError as error:
+        logger.warning(f"Could not remove the move's temp file {path}: {error}")
+        return False
+
+
+# ── removing ─────────────────────────────────────────────────────────
+
+
+def _in_use(error):
+    return getattr(error, "winerror", None) in _WINDOWS_IN_USE
+
+
+def remove_file(path, *, sleep=time.sleep):
+    """Remove one file; a file another program holds open (Windows) is
+    retried 3 times. Returns True when it is gone."""
+    for wait in (0.0,) + REMOVE_RETRIES:
+        if wait:
+            sleep(wait)
+        try:
+            os.unlink(path)
+            return True
+        except FileNotFoundError:
+            return True
+        except OSError as error:
+            if not _in_use(error):
+                return False
+    return False
+
+
+def remove_empty_folders(top, *, keep_top=False):
+    """``rmdir`` the empty folders below *top* (and *top*), bottom-up, never
+    through a link. A folder holding only OS metadata files loses them first.
+    Returns the folders that could not be removed."""
+    top = Path(top)
+    left = []
+    if not top.is_dir() or preview_location.is_link_like(top):
+        return left
+    # Top-down, pruning every link, junction or reparse point, then bottom-up:
+    # never into a folder outside *top* (Windows junctions included).
+    folders = []
+    for folder, dirs, _files in os.walk(top):
+        folder = Path(folder)
+        dirs[:] = [d for d in dirs if not preview_location.is_link_like(folder / d)]
+        folders.append(folder)
+    for folder in reversed(folders):
+        if folder == top and keep_top:
+            continue
+        try:
+            names = os.listdir(folder)
+        except OSError:
+            left.append(folder)
+            continue
+        if names and all(is_os_metadata(name) for name in names):
+            for name in names:
+                remove_file(folder / name)
+        try:
+            os.rmdir(folder)
+        except OSError:
+            left.append(folder)
+    return left

py/api/thumbnail_move_plan.py

@@ -0,0 +1,492 @@
+"""Move thumbnails, the review (#320 Part 3 phase 1): what a move would copy,
+remove and leave, checked before anything is written."""
+
+import hashlib
+import json
+import os
+import secrets
+import shutil
+import time
+import uuid
+from dataclasses import dataclass, field
+from pathlib import Path
+
+from aiohttp import web
+
+from . import preview_location, preview_store
+from . import thumbnail_move_files as files
+from .job_routes import busy_response
+from .jobs import THUMBNAILS_JOB_KIND, JobBusy
+from .thumbnail_location import default_path
+
+try:
+    from ...utils import library_location
+except ImportError:  # tests import the package top-level
+    from utils import library_location
+
+MOVE_TITLE = "Move thumbnails"
+FINISHING_TITLE = "Move thumbnails (finishing)"
+PRUNE_TITLE = "Remove thumbnails of removed sources"
+MOVE_GROUPS = ("moved", "skipped", "left", "errors")
+PRUNE_GROUPS = ("removed", "left", "errors")
+PLAN_TTL = 600.0
+JOURNAL_NAME = "thumbnail-move.json"
+SPACE_MARGIN = 64 * 2**20
+SPACE_SHARE = 0.05
+LISTED = 5
+SENTENCE = (
+    "Every thumbnail is copied and checked before Prompt Manager switches. "
+    "Scans, deletes and tagging wait until it finishes; closing this page does "
+    "not stop it."
+)
+
+
+def _number(value):
+    return f"{value:,}"
+
+
+def _plural(count, one, many=None):
+    return f"{_number(count)} {one if count == 1 else (many or one + 's')}"
+
+
+def _size(value):
+    size = float(value)
+    for unit in ("B", "KB", "MB", "GB"):
+        if size < 1024 or unit == "GB":
+            return f"{size:.1f} {unit}" if unit != "B" else f"{int(size)} B"
+        size /= 1024
+    return f"{size:.1f} GB"
+
+
+def _refusal(code, status=None, **extra):
+    known_status, text = preview_location.MESSAGES.get(code, (400, code))
+    return web.json_response(
+        {"success": False, "code": code, "error": text, **extra},
+        status=status or known_status,
+    )
+
+
+def journal_path():
+    """``<data folder>/data/thumbnail-move.json`` (only while a move copies)."""
+    return library_location.cache_dir(JOURNAL_NAME)
+
+
+def _read_json(path):
+    try:
+        with open(path, "r", encoding="utf-8") as handle:
+            data = json.load(handle)
+    except (OSError, ValueError):
+        return None
+    return data if isinstance(data, dict) else None
+
+
+def _write_json(path, data):
+    os.makedirs(os.path.dirname(path), exist_ok=True)
+    temp = f"{path}.{os.getpid()}.tmp"
+    with open(temp, "w", encoding="utf-8") as handle:
+        json.dump(data, handle, indent=2)
+        handle.flush()
+        os.fsync(handle.fileno())
+    os.replace(temp, path)
+
+
+def _remove(path):
+    try:
+        os.unlink(path)
+    except FileNotFoundError:
+        pass  # already gone: removing it is the goal
+
+
+def _is_network(path):
+    text = str(path)
+    if text.startswith(("\\\\", "//")):
+        return True
+    if os.name == "nt":
+        try:
+            import ctypes
+
+            drive = os.path.splitdrive(os.path.abspath(text))[0] + "\\"
+            return ctypes.windll.kernel32.GetDriveTypeW(drive) == 4  # DRIVE_REMOTE
+        except (AttributeError, OSError):
+            return False
+    return False
+
+
+def _volume(path):
+    current = Path(os.path.abspath(path))
+    while not current.exists() and current.parent != current:
+        current = current.parent
+    try:
+        return os.stat(current).st_dev
+    except OSError:
+        return None
+
+
+def _free(path):
+    current = Path(os.path.abspath(path))
+    while not current.exists() and current.parent != current:
+        current = current.parent
+    try:
+        return shutil.disk_usage(current).free
+    except OSError:
+        return None
+
+
+# ── the plan ─────────────────────────────────────────────────────────
+
+
+@dataclass
+class SourcePlan:
+    """One listed source: its old store, its new store and what moves."""
+
+    root: str
+    label: str
+    key: str | None
+    old_store: str | None
+    new_store: str
+    scan: files.StoreScan | None
+    available: bool = True
+    too_long: list = field(default_factory=list)
+
+
+@dataclass
+class MovePlan:
+    """A reviewed move: from, to, the sources and the removals."""
+
+    token: str
+    created: float
+    revision: str
+    old: preview_location.ParsedLocation
+    target: dict  # {"mode", "path"}
+    target_id: str | None
+    target_exists: bool
+    target_default: bool
+    install: str
+    sources: list
+    unlisted: list  # (key, store path, scan)
+    needs_bytes: int
+    confirm_required: bool
+
+    @property
+    def to_folder(self):
+        return self.target["mode"] == preview_store.FOLDER_MODE
+
+    @property
+    def from_folder(self):
+        return self.old.mode == preview_store.FOLDER_MODE
+
+
+def _roots_revision(roots):
+    text = json.dumps(sorted(str(root) for root in roots))
+    return hashlib.sha256(text.encode("utf-8")).hexdigest()[:16]
+
+
+class ThumbnailMovePlanMixin:
+    """The move's review and its routes."""
+
+    _move_plans = None
+
+    # ── routes ────────────────────────────────────────────────────────
+
+    def _register_thumbnail_move_routes(self, routes):
+        @routes.post("/prompt_manager/images/thumbnails/plan")
+        async def thumbnail_move_plan_route(request):
+            return await self.plan_thumbnail_move(request)
+
+        @routes.post("/prompt_manager/images/thumbnails/location/recreate")
+        async def thumbnail_location_recreate_route(request):
+            return await self.recreate_thumbnail_folder(request)
+
+        @routes.post("/prompt_manager/images/thumbnails/location/adopt")
+        async def thumbnail_location_adopt_route(request):
+            return await self.adopt_thumbnail_folder(request)
+
+    # ── review ────────────────────────────────────────────────────────
+
+    async def plan_thumbnail_move(self, request):
+        try:
+            body = await request.json()
+        except (ValueError, UnicodeDecodeError):
+            body = None
+        if not isinstance(body, dict) or body.get("mode") != "move":
+            return web.json_response(
+                {
+                    "success": False,
+                    "code": "invalid-request",
+                    "error": "mode must be move",
+                },
+                status=400,
+            )
+        target = body.get("location")
+        if not isinstance(target, dict) or target.get("mode") not in (
+            preview_store.SOURCE_MODE,
+            preview_store.FOLDER_MODE,
+        ):
+            return web.json_response(
+                {
+                    "success": False,
+                    "code": "invalid-request",
+                    "error": "location must name a mode",
+                },
+                status=400,
+            )
+        blocker = self._jobs.blocker(THUMBNAILS_JOB_KIND, admission="guarded")
+        if blocker is not None:
+            return busy_response(JobBusy(blocker))
+        try:
+            outcome = await self._run_in_executor(self._build_move_plan, target)
+        except Exception:
+            self.logger.exception("The thumbnail move could not be planned")
+            return web.json_response(
+                {
+                    "success": False,
+                    "code": None,
+                    "error": "The move could not be planned",
+                },
+                status=500,
+            )
+        if isinstance(outcome, str):
+            return _refusal(outcome)
+        plan, payload = outcome
+        if payload.get("insufficient"):
+            return _refusal(
+                "insufficient-space",
+                status=507,
+                error_detail=payload["insufficient"],
+                plan=payload,
+            )
+        self._store_move_plan(plan)
+        return web.json_response({"success": True, **payload})
+
+    def _store_move_plan(self, plan):
+        if self._move_plans is None:
+            self._move_plans = {}
+        now = time.monotonic()
+        self._move_plans = {
+            token: kept
+            for token, kept in self._move_plans.items()
+            if now - kept.created < PLAN_TTL
+        }
+        self._move_plans[plan.token] = plan
+
+    def _build_move_plan(self, target):
+        """``(plan, payload)`` or a refusal code (blocking)."""
+        old = preview_store.setting()
+        location = preview_store.current()
+        install = old.install or str(uuid.uuid4())
+        target_path = None
+        default = default_path()
+        if target["mode"] == preview_store.FOLDER_MODE:
+            raw = target.get("path")
+            if not isinstance(raw, str):
+                return "location-not-absolute"
+            context = self._thumbnail_location_context()
+            code = preview_location.location_problem(
+                raw, purpose="move", context=context
+            )
+            if code is not None:
+                return code
+            target_path = os.path.abspath(raw)
+        elif old.mode == preview_store.SOURCE_MODE and not old.problem:
+            return "location-same"
+        target_default = bool(default) and os.path.normcase(
+            target_path or ""
+        ) == os.path.normcase(os.path.abspath(default))
+        target_exists = bool(target_path) and os.path.isdir(target_path)
+        target_marker = (
+            preview_location.read_marker(target_path) if target_exists else None
+        )
+        roots = [str(root) for root in self._get_all_output_dirs()]
+        labels = self._source_labels(roots)
+        old_keys = {}
+        if old.mode == preview_store.FOLDER_MODE and location.available:
+            marker = preview_location.read_marker(old.path) or {}
+            old_keys = marker.get("keys") or {}
+        new_keys = dict((target_marker or {}).get("keys") or {})
+        taken = {key: entry.get("identity") for key, entry in new_keys.items()}
+        sources = []
+        for root in roots:
+            identity = os.path.normcase(os.path.realpath(root))
+            dev_ino = preview_location.file_identity(root)
+            old_key = None
+            old_store = None
+            available = True
+            if old.mode == preview_store.SOURCE_MODE:
+                old_store = str(preview_store.source_folder(root))
+            elif location.available:
+                old_key = preview_location.find_key(old_keys, root, identity, dev_ino)
+                if old_key:
+                    old_store = os.path.join(old.path, old_key)
+            else:
+                available = False
+            key = None
+            if target_path:
+                key = old_key or preview_location.find_key(
+                    new_keys, root, identity, dev_ino
+                )
+                if key is None or taken.get(key, identity) != identity:
+                    key = preview_location.source_key(root, identity, taken)
+                taken[key] = identity
+                new_store = os.path.join(target_path, key)
+            else:
+                new_store = str(preview_store.source_folder(root))
+            scan = None
+            if old_store:
+                scan = files.scan_store(old_store, root)
+            plan = SourcePlan(
+                root, labels.get(root, root), key, old_store, new_store, scan, available
+            )
+            if scan is not None and target_path:
+                plan.too_long = [
+                    item[0]
+                    for item in scan.copy
+                    if files.too_long(Path(new_store).joinpath(*item[0]))
+                ]
+            sources.append(plan)
+        unlisted = self._unlisted_keys(old, location, old_keys, install)
+        needs = sum(source.scan.bytes for source in sources if source.scan is not None)
+        plan = MovePlan(
+            token=secrets.token_urlsafe(16),
+            created=time.monotonic(),
+            revision=_roots_revision(roots),
+            old=old,
+            target={"mode": target["mode"], "path": target_path},
+            target_id=(target_marker or {}).get("id"),
+            target_exists=target_exists,
+            target_default=target_default,
+            install=install,
+            sources=sources,
+            unlisted=unlisted,
+            needs_bytes=needs,
+            confirm_required=bool(target_path) and not target_default,
+        )
+        return plan, self._plan_payload(plan, location)
+
+    def _unlisted_keys(self, old, location, keys, install):
+        """This install's keys whose source no longer is in any root list
+        (``(key, store, scan)``), in the folder in use."""
+        if old.mode != preview_store.FOLDER_MODE or not location.available:
+            return []
+        listed = {
+            os.path.normcase(os.path.realpath(root))
+            for root in self._all_source_roots()
+        }
+        found = []
+        for key, entry in sorted(keys.items()):
+            if not preview_location.valid_key(key) or not isinstance(entry, dict):
+                continue  # a marker is data: never a path out of the folder
+            if entry.get("install") not in (install, None):
+                continue
+            if entry.get("identity") in listed:
+                continue
+            store = os.path.join(old.path, key)
+            found.append(
+                (key, store, files.scan_store(store, entry.get("path") or store))
+            )
+        return found
+
+    def _source_labels(self, roots):
+        from . import sources
+
+        labels = sources._comfyui_labels()
+        return {root: sources.source_label(root, labels) for root in roots}
+
+    def _plan_payload(self, plan, location):
+        sources = plan.sources
+        copies = sum(len(s.scan.copy) for s in sources if s.scan)
+        orphans = sum(len(s.scan.orphans) for s in sources if s.scan)
+        temps = sum(len(s.scan.temps) for s in sources if s.scan)
+        links = sum(s.scan.links for s in sources if s.scan)
+        left = [
+            f"{s.label} · {'/'.join(item)}"
+            for s in sources
+            if s.scan
+            for item in s.scan.left
+        ]
+        too_long = [
+            f"{s.label} · {'/'.join(item)}" for s in sources for item in s.too_long
+        ]
+        unlisted_count = sum(
+            len(scan.copy) + len(scan.orphans) for _k, _s, scan in plan.unlisted
+        )
+        free = None
+        insufficient = None
+        if plan.needs_bytes:
+            volumes = {}
+            for source in sources:
+                if source.scan is None or not source.scan.bytes:
+                    continue
+                volume = _volume(source.new_store)
+                entry = volumes.setdefault(volume, [source.new_store, 0])
+                entry[1] += source.scan.bytes
+            for where, needed in volumes.values():
+                available = _free(where)
+                if available is None:
+                    continue
+                free = available if free is None else min(free, available)
+                margin = max(SPACE_MARGIN, int(needed * SPACE_SHARE))
+                if needed + margin > available:
+                    insufficient = {
+                        "needed": needed + margin,
+                        "free": available,
+                        "text": (
+                            f"There is not enough space for the thumbnails: "
+                            f"{_size(available)} free, {_size(needed + margin)} "
+                            "needed. Free some space or choose another folder."
+                        ),
+                    }
+        payload = {
+            "mode": "move",
+            "from": {
+                "mode": plan.old.mode,
+                "path": plan.old.path,
+                "available": location.available,
+            },
+            "to": {
+                "mode": plan.target["mode"],
+                "path": plan.target["path"],
+                "exists": plan.target_exists,
+                "will_create": bool(plan.target["path"]) and not plan.target_exists,
+                "default": plan.target_default,
+            },
+            "sources": [
+                {
+                    "label": s.label,
+                    "key": s.key,
+                    "files": len(s.scan.copy) if s.scan else 0,
+                    "bytes": s.scan.bytes if s.scan else 0,
+                    "too_long": len(s.too_long),
+                    "links": s.scan.links if s.scan else 0,
+                    "orphans": len(s.scan.orphans) if s.scan else 0,
+                    "left_in_place": len(s.scan.left) if s.scan else 0,
+                    "available": s.available,
+                }
+                for s in sources
+            ],
+            "totals": {
+                "sources": len(sources),
+                "files": copies,
+                "bytes": plan.needs_bytes,
+                "orphans": orphans,
+                "temps": temps,
+                "links": links,
+                "left_in_place": len(left),
+                "too_long": len(too_long),
+                "unlisted_sources": len(plan.unlisted),
+                "unlisted_files": unlisted_count,
+            },
+            "too_long_listed": too_long[:LISTED],
+            "left_listed": left[:LISTED],
+            "unlisted": [
+                {"key": key, "files": len(scan.copy) + len(scan.orphans)}
+                for key, _store, scan in plan.unlisted
+            ],
+            "needs_bytes": plan.needs_bytes,
+            "free_bytes": free,
+            "confirm_required": plan.confirm_required,
+            "token": plan.token,
+            "sentence": SENTENCE,
+        }
+        if insufficient:
+            payload["insufficient"] = insufficient
+        return payload

py/api/thumbnail_move_recovery.py

@@ -0,0 +1,246 @@
+"""Move thumbnails after a restart and the unavailable folder's actions
+(#320 Parts 3 and 4): undo an interrupted copy, finish a removal that
+waited, Recreate folder and Use this folder."""
+
+import os
+import threading
+from pathlib import Path
+
+from aiohttp import web
+
+from . import preview_location, preview_store
+from . import thumbnail_move_files as files
+from .job_routes import busy_response
+from .jobs import THUMBNAILS_JOB_KIND, JobBusy
+from .thumbnail_move_plan import (
+    FINISHING_TITLE,
+    MOVE_GROUPS,
+    _read_json,
+    _refusal,
+    _remove,
+)
+from . import thumbnail_move_plan as move_plan
+
+
+class ThumbnailMoveRecoveryMixin:
+    """Startup recovery, the finishing job, Recreate and Use this folder."""
+
+    _move_recover_lock = threading.Lock()
+    # Per API: set once recover_thumbnail_move has run (a move waits for it).
+    _move_recovered = False
+
+    # ── startup ───────────────────────────────────────────────────────
+
+    def recover_thumbnail_move(self):
+        """Startup (B4): undo an interrupted copy, or forget a journal the
+        switch already made true. Idempotent; the move start waits for it."""
+        with self._move_recover_lock:
+            if self._move_recovered:
+                return
+            try:
+                self._recover_journal()
+            except Exception:
+                self.logger.warning(
+                    "The interrupted thumbnail move check failed", exc_info=True
+                )
+            self._move_recovered = True
+
+    def _recover_journal(self):
+        path = move_plan.journal_path()
+        journal = _read_json(path)
+        if journal is None:
+            return
+        setting = preview_store.setting()
+        target = journal.get("to") or {}
+        now = setting.path if setting.mode == preview_store.FOLDER_MODE else None
+        if setting.mode == target.get("mode") and (
+            now is None
+            or os.path.normcase(now) == os.path.normcase(target.get("path") or "")
+        ):
+            _remove(path)  # the switch happened; only the journal was left
+            return
+        for store in journal.get("stores") or []:
+            if not isinstance(store, str) or not os.path.isdir(store):
+                continue
+            for current, dirs, names in os.walk(store):
+                dirs[:] = [
+                    d
+                    for d in dirs
+                    if not preview_location.is_link_like(Path(current) / d)
+                ]
+                for name in names:
+                    if files.MOVE_TEMP.search(name):
+                        files.remove_file(Path(current) / name)
+        for folder in reversed(journal.get("created_dirs") or []):
+            if not isinstance(folder, str) or not os.path.isdir(folder):
+                continue
+            for current, dirs, names in os.walk(folder):
+                dirs[:] = [
+                    d
+                    for d in dirs
+                    if not preview_location.is_link_like(Path(current) / d)
+                ]
+                for name in names:
+                    if files._preview_parts(name) or files.is_temp_name(name):
+                        files.remove_file(Path(current) / name)
+            files.remove_empty_folders(folder)
+        folder = target.get("path")
+        if journal.get("created_folder") and folder and os.path.isdir(folder):
+            names = set(os.listdir(folder))
+            if names <= {preview_location.MARKER_NAME}:
+                _remove(preview_location.marker_path(folder))
+                files.remove_empty_folders(folder)
+        _remove(path)
+        self.logger.info("An interrupted thumbnail move was undone; nothing changed")
+
+    def _cleanup_reachable(self, entry):
+        folder = entry.get("folder")
+        if folder:
+            return os.path.isdir(folder)
+        return True
+
+    async def finish_thumbnail_move(self, _app=None):
+        """Startup: recover, then start "Move thumbnails (finishing)" when old
+        copies wait and every one is reachable (on the server's loop)."""
+        await self._run_in_executor(self.recover_thumbnail_move)
+        cleanup = list(preview_store.setting().cleanup)
+        if not cleanup:
+            return None
+        reachable = await self._run_in_executor(
+            lambda: all(self._cleanup_reachable(entry) for entry in cleanup)
+        )
+        if not reachable:
+            return None
+        try:
+            return self._jobs.start(
+                THUMBNAILS_JOB_KIND,
+                FINISHING_TITLE,
+                admission="guarded",
+                cancellable=False,
+                scope={"mode": "move", "phase": "remove"},
+                groups=MOVE_GROUPS,
+                runner=self._run_finishing,
+            )
+        except JobBusy:
+            return None
+
+    async def _run_finishing(self, job):
+        await self._remove_old_copies(job, None)
+        left = len(preview_store.setting().cleanup)
+        self._end_move(
+            job,
+            "finished",
+            (
+                "Old thumbnails removed."
+                if not left
+                else (
+                    "Some old thumbnails are still in use; they are removed at "
+                    "the next start."
+                )
+            ),
+        )
+
+    # ── Recreate and Use this folder (Part 4) ─────────────────────────
+
+    async def recreate_thumbnail_folder(self, _request):
+        blocker = self._jobs.blocker(THUMBNAILS_JOB_KIND, admission="guarded")
+        if blocker is not None:
+            return busy_response(JobBusy(blocker))
+        code = await self._run_in_executor(self._recreate_folder)
+        if code is not None:
+            status = (
+                409 if code in ("location-has-marker", "location-not-empty") else 400
+            )
+            if code == "location-has-marker":
+                return web.json_response(
+                    {
+                        "success": False,
+                        "code": code,
+                        "error": "The folder is back with its thumbnails. Nothing "
+                        "was recreated.",
+                    },
+                    status=status,
+                )
+            return _refusal(code, status=status)
+        return web.json_response(
+            {
+                "success": True,
+                "location": self._thumbnail_location_payload(with_path=True),
+            }
+        )
+
+    def _recreate_folder(self):
+        setting = preview_store.setting()
+        if setting.mode != preview_store.FOLDER_MODE or not setting.path:
+            return "location-same"
+        folder = setting.path
+        if os.path.isdir(folder):
+            if preview_location.read_marker(folder) is not None:
+                return "location-has-marker"
+            if not preview_location.holds_only_metadata(folder):
+                return "location-not-empty"
+        elif not os.path.isdir(os.path.dirname(folder)):
+            return "location-parent-missing"
+        else:
+            try:
+                os.mkdir(folder)
+            except OSError:
+                return "location-not-writable"
+        try:
+            marker_id = preview_location.init_folder(folder, created_by_pm=True)
+        except OSError:
+            return "location-not-writable"
+        return self._commit_location_id(marker_id)
+
+    async def adopt_thumbnail_folder(self, _request):
+        blocker = self._jobs.blocker(THUMBNAILS_JOB_KIND, admission="guarded")
+        if blocker is not None:
+            return busy_response(JobBusy(blocker))
+        code = await self._run_in_executor(self._adopt_folder)
+        if code is not None:
+            return _refusal(code)
+        return web.json_response(
+            {
+                "success": True,
+                "location": self._thumbnail_location_payload(with_path=True),
+            }
+        )
+
+    def _adopt_folder(self):
+        setting = preview_store.setting()
+        if setting.mode != preview_store.FOLDER_MODE or not setting.path:
+            return "location-same"
+        try:
+            marker = preview_location.read_marker(setting.path)
+        except OSError:
+            return "location-unavailable"
+        if marker is None:
+            return "location-unavailable"
+        return self._commit_location_id(marker["id"])
+
+    def _commit_location_id(self, marker_id):
+        from ..config import PromptManagerConfig
+        from ..config_store import get_config_store
+
+        value = dict(preview_store.setting().to_config())
+        value["id"] = marker_id
+
+        def mutate(config):
+            section = config.get("thumbnails")
+            if not isinstance(section, dict):
+                section = {}
+            section["location"] = value
+            config["thumbnails"] = section
+            return config
+
+        def publish(_config):
+            PromptManagerConfig.THUMBNAIL_LOCATION = value
+            preview_store.configure(value, check=False)
+
+        try:
+            get_config_store().commit(mutate, expected_revision=None, publish=publish)
+        except Exception:
+            self.logger.exception("The thumbnail folder could not be saved")
+            return "location-not-writable"
+        preview_store.refresh()
+        return None

py/api/thumbnail_prune.py

@@ -0,0 +1,175 @@
+"""Remove thumbnails of removed sources (#320 Part 5, D200): a Thumbnail
+maintenance scope, offered only in folder mode (31A).
+
+*Removed* means a key of this install in the thumbnail folder's marker whose
+identity matches no root in any root list. The job removes the
+preview-grammar files and temps of those key folders, ``rmdir``s the empty
+folders, drops the keys from the marker and lists anything else as "Left in
+place". Kind ``thumbnails``, ``mode: "prune"``.
+"""
+
+import functools
+import os
+from pathlib import Path
+
+from aiohttp import web
+
+from . import preview_location, preview_store
+from . import thumbnail_move_files as files
+from .jobs import THUMBNAILS_JOB_KIND
+from .thumbnail_location import location_refusal
+from .thumbnail_move_plan import _number, _plural, _size
+
+PRUNE_TITLE = "Remove thumbnails of removed sources"
+PRUNE_GROUPS = ("removed", "left", "errors")
+
+
+def _prune_line(count, files_count, size):
+    if not count:
+        return "No sources were removed: nothing to remove."
+    return (
+        f"{_plural(count, 'source')} no longer listed: "
+        f"{_plural(files_count, 'thumbnail')} ({_size(size)}) are removed."
+    )
+
+
+class ThumbnailPruneMixin:
+    """The prune plan and job."""
+
+    def _prune_targets(self):
+        """``[(key, store, scan)]`` of this install's unlisted keys (blocking)."""
+        setting = preview_store.setting()
+        location = preview_store.current()
+        if setting.mode != preview_store.FOLDER_MODE or not location.available:
+            return []
+        marker = preview_location.read_marker(setting.path) or {}
+        install = setting.install
+        return self._unlisted_keys(setting, location, marker.get("keys") or {}, install)
+
+    def _prune_payload(self, targets):
+        files_count = sum(
+            len(scan.copy) + len(scan.orphans) for _k, _s, scan in targets
+        )
+        size = sum(scan.bytes for _k, _s, scan in targets)
+        return {
+            "success": True,
+            "mode": "prune",
+            "sources": [
+                {
+                    "key": key,
+                    "files": len(scan.copy) + len(scan.orphans),
+                    "bytes": scan.bytes,
+                    "left_in_place": len(scan.left),
+                }
+                for key, _store, scan in targets
+            ],
+            "targets": files_count,
+            "bytes": size,
+            "line": _prune_line(len(targets), files_count, size),
+            "folder": preview_store.setting().path,
+        }
+
+    async def get_prune_plan(self):
+        refused = location_refusal()
+        if refused is not None:
+            return refused
+        if preview_store.current().mode != preview_store.FOLDER_MODE:
+            return web.json_response(
+                {
+                    "success": False,
+                    "code": "invalid-request",
+                    "error": "Thumbnails are kept inside each source: no removed "
+                    "source has thumbnails here.",
+                },
+                status=400,
+            )
+        targets = await self._run_in_executor(self._prune_targets)
+        return web.json_response(
+            self._prune_payload(targets), headers={"Cache-Control": "no-store"}
+        )
+
+    def start_thumbnail_prune(self):
+        """Start the prune job (sync, on the loop); raises JobBusy."""
+        return self._jobs.start(
+            THUMBNAILS_JOB_KIND,
+            PRUNE_TITLE,
+            admission="guarded",
+            cancellable=True,
+            scope={"mode": "prune"},
+            groups=PRUNE_GROUPS,
+            runner=self._run_thumbnail_prune,
+        )
+
+    async def _run_thumbnail_prune(self, job):
+        targets = await self._run_in_executor(self._prune_targets)
+        job.update(total=len(targets), processed=0)
+        setting = preview_store.setting()
+        for done, (key, store, scan) in enumerate(targets, 1):
+            if job.cancelled:
+                break
+            removed, failed = await self._run_in_executor(
+                functools.partial(self._prune_store, store)
+            )
+            job.record_counted(
+                "removed", removed, [{"path": key, "detail": "Thumbnails removed"}]
+            )
+            if failed:
+                job.record_counted("errors", failed)
+            if scan.left:
+                job.record_counted(
+                    "left",
+                    len(scan.left),
+                    [{"path": f"{key}/{'/'.join(item)}"} for item in scan.left],
+                )
+            await self._run_in_executor(
+                functools.partial(self._forget_key, setting.path, key, store)
+            )
+            job.update(
+                processed=done,
+                progress=int(done * 100 / len(targets)),
+                status_line=f"Removing thumbnails of {_number(done)} of "
+                f"{_number(len(targets))} removed sources",
+            )
+            job.publish({"type": "progress", "summary": job.summary()})
+        line = f"{_number(job.counters.get('removed', 0))} thumbnails removed"
+        if job.counters.get("errors"):
+            line += f" · {_plural(job.counters['errors'], 'error')}"
+        job.update(status_line=line, progress=100)
+        job.log_line(line)
+        job.finish("cancelled" if job.cancelled else "finished")
+
+    @staticmethod
+    def _prune_store(store):
+        """Remove the previews and temps of one key folder; ``(removed, failed)``."""
+        removed = failed = 0
+        store = Path(store)
+        if not store.is_dir() or preview_location.is_link_like(store):
+            return 0, 0
+        for current, dirs, names in os.walk(store):
+            dirs[:] = [
+                d for d in dirs if not preview_location.is_link_like(Path(current) / d)
+            ]
+            for name in names:
+                path = Path(current) / name
+                if not (files.is_temp_name(name) or files._preview_parts(name)):
+                    continue
+                if preview_location.is_link_like(path):
+                    continue
+                if files.remove_file(path):
+                    removed += 1
+                else:
+                    failed += 1
+        files.remove_empty_folders(store)
+        return removed, failed
+
+    @staticmethod
+    def _forget_key(folder, key, store):
+        """Drop *key* from the marker once its folder is gone."""
+        if os.path.exists(store):
+            return
+
+        def forget(marker):
+            marker["keys"].pop(key, None)
+
+        with preview_store.write_lock:
+            preview_location.write_marker(folder, forget)

tests/browser/system-concept.spec.mjs

@@ -656,9 +656,10 @@ function mapSourcesConcept(reference, { settings, sources }) {
             // D113: the range ends at the server's core count (the concept's fixed 16) and shows the real worker count.
             return [{ ...row, control: { ...row.control, max: String(settings.cpu_count), value: String(settings.worker_threads) } }];
         }),
-        // D158: the inserted row after the first setting row; D98: "Refresh files" closes the surface.
+        // D158: the inserted row after the first setting row; D196 (#320): the Thumbnail location block after
+        // "Thumbnail generation"; D98: "Refresh files" closes the surface.
         surface: { ...reference.surface, tree: [...reference.surface.tree.slice(0, rowAt + 1), 'div.setting-row',
-            ...reference.surface.tree.slice(rowAt + 1), 'div.operation'] },
+            ...reference.surface.tree.slice(rowAt + 1), 'div.setting-block', 'div.operation'] },
         operations: [...reference.operations, { ...reference.operations[0], title: REFRESH_FILES.title,
             description: REFRESH_FILES.description, actions: [{ ...reference.operations[0].actions[0], text: REFRESH_FILES.label }] }],
     };

tests/browser/thumbnail_location_fixture.py

@@ -31,12 +31,43 @@ def register_thumbnail_location_commands(
 ) -> Callable[[], None]:
     """Register the commands; returns the reset every reset runs."""
     from py.api import preview_location, preview_store
+    from py.api import thumbnail_location, thumbnail_move_plan
+    from py.config import PromptManagerConfig
+    from py.config_store import get_config_store
 
     folder = root / FOLDER_NAME
+    default = root / "default-thumbs"
+    custom = root / "custom-thumbs"  # a folder a page types in
+    journal = root / "data" / "thumbnail-move.json"
+    # The move's journal and default folder live in this server's root, never
+    # in the checkout's data folder.
+    thumbnail_move_plan.journal_path = lambda: str(journal)
+    thumbnail_move_plan.default_path = lambda: str(default)
+    thumbnail_location.default_path = lambda: str(default)
+    api.recover_thumbnail_move()
+
+    def forget_config(config: dict) -> dict:
+        section = config.get("thumbnails")
+        if isinstance(section, dict):
+            section.pop("location", None)
+        return config
+
+    real_free = thumbnail_move_plan._free
 
     def reset() -> None:
+        thumbnail_move_plan._free = real_free
         preview_store.configure(None)
-        shutil.rmtree(folder, ignore_errors=True)
+        PromptManagerConfig.THUMBNAIL_LOCATION = None
+        store = get_config_store()
+        config, _revision = store.read()
+        if (
+            isinstance(config.get("thumbnails"), dict)
+            and "location" in config["thumbnails"]
+        ):
+            store.commit(forget_config, expected_revision=None)
+        for path in (folder, default, custom):
+            shutil.rmtree(path, ignore_errors=True)
+        journal.unlink(missing_ok=True)
 
     def use(mode: str) -> dict:
         reset()
@@ -63,14 +94,18 @@ def register_thumbnail_location_commands(
         return web.json_response(await api._run_in_executor(use, mode))
 
     def files() -> dict:
+        setting = preview_store.setting()
+        where = Path(setting.path) if setting.path else folder
         previews = sorted(
-            path.relative_to(folder).as_posix()
-            for path in folder.rglob("*_thumb.*")
+            path.relative_to(where).as_posix()
+            for path in where.rglob("*_thumb.*")
             if path.is_file()
         )
         sources = [str(path) for path in api._get_all_output_dirs()]
         return {
             "folder": previews,
+            "folder_exists": folder.exists() or default.exists() or custom.exists(),
+            "custom_path": str(custom),
             "source_thumbnails": any(
                 (Path(source) / "thumbnails").exists() for source in sources
             ),
@@ -81,7 +116,8 @@ def register_thumbnail_location_commands(
         return web.json_response(await api._run_in_executor(files))
 
     def remove() -> bool:
-        shutil.rmtree(folder, ignore_errors=True)
+        setting = preview_store.setting()
+        shutil.rmtree(setting.path or folder, ignore_errors=True)
         return preview_store.refresh()
 
     @routes.post("/__test__/thumbnail-folder/remove")
@@ -89,4 +125,46 @@ def register_thumbnail_location_commands(
         available = await api._run_in_executor(remove)
         return web.json_response({"available": available})
 
+    def foreign() -> bool:
+        """Another install recreated the folder: its marker has another id."""
+        setting = preview_store.setting()
+
+        def other(marker: dict) -> None:
+            marker["id"] = "another-install"
+
+        preview_location.write_marker(setting.path, other)
+        return preview_store.refresh()
+
+    @routes.post("/__test__/thumbnail-folder/foreign")
+    async def foreign_folder(_request: web.Request) -> web.Response:
+        return web.json_response({"available": await api._run_in_executor(foreign)})
+
+    def unlisted() -> None:
+        """A source this install no longer lists left its thumbnails here."""
+        setting = preview_store.setting()
+
+        def add(marker: dict) -> None:
+            marker["keys"]["gone-0123abcd"] = {
+                "path": "/gone",
+                "identity": "/gone",
+                "install": setting.install,
+            }
+
+        preview_location.write_marker(setting.path, add)
+        gone = Path(setting.path) / "gone-0123abcd"
+        gone.mkdir()
+        (gone / "x_thumb.webp").write_bytes(b"preview")
+        preview_store.refresh()
+
+    @routes.post("/__test__/thumbnail-folder/unlisted")
+    async def unlisted_source(_request: web.Request) -> web.Response:
+        await api._run_in_executor(unlisted)
+        return web.json_response({"success": True})
+
+    @routes.post("/__test__/thumbnail-space/full")
+    async def space_full(_request: web.Request) -> web.Response:
+        """Every target disk reports 1 byte free until the next reset."""
+        thumbnail_move_plan._free = lambda _path: 1
+        return web.json_response({"success": True})
+
     return reset

tests/browser/workspace-thumbnail-location.spec.mjs

@@ -0,0 +1,213 @@
+import { startCoverage, saveCoverage } from '../../tools/vendor/browser-coverage.mjs';
+import { test, expect } from './fixtures.mjs';
+import { seedConcept } from './concept-inventory.mjs';
+
+// #320 Thumbnail location (D196–D201, the approved 320 mock) on the concept-shaped real fixture: the row in System ›
+// Sources & scans reviews and runs Move thumbnails both ways, the review refuses a folder in place, and an unplugged
+// thumbnail folder shows the notice whose Recreate folder brings it back. The hooks are thumbnail_location_fixture.py's.
+const workspace = '/prompt_manager/workspace';
+const NOTICE_KEY = 'pm_update_notice_dismissed';
+
+test.beforeEach(async ({ page, request, browserName }) => {
+    await startCoverage(page, browserName);
+    await seedConcept(request);
+    await expect(await request.post('/__test__/system/reset')).toBeOK();
+    await expect(await request.post('/__test__/thumbnail-location', { data: { mode: 'source' } })).toBeOK();
+    page.locationErrors = [];
+    page.on('pageerror', error => page.locationErrors.push(error.message));
+    await page.addInitScript(key => { if (location.protocol.startsWith('http')) localStorage.setItem(key, '3.2.0'); }, NOTICE_KEY);
+});
+
+test.afterEach(async ({ page, request, browserName }) => {
+    await saveCoverage(page, browserName);
+    await request.post('/__test__/thumbnail-location', { data: { mode: 'source' } });
+    expect(page.locationErrors).toEqual([]);
+});
+
+const block = page => page.locator('#location-block');
+const dialog = page => page.locator('dialog#modal[open]');
+
+async function files(request) {
+    const response = await request.get('/__test__/thumbnail-files');
+    await expect(response).toBeOK();
+    return response.json();
+}
+
+async function openSources(page) {
+    await page.goto(`${workspace}#/system/sources`);
+    await expect(block(page).locator('#location-status')).not.toHaveText('');
+}
+
+/** Review the row's choice and start the move; resolves once its job view says it finished. */
+async function moveFromRow(page) {
+    const planned = page.waitForResponse(response => response.url().endsWith('/images/thumbnails/plan')
+        && response.request().method() === 'POST');
+    await block(page).getByRole('button', { name: 'Review move…' }).click();
+    expect((await planned).status()).toBe(200);
+    await expect(dialog(page).locator('h2')).toHaveText('Move thumbnails');
+    await dialog(page).getByRole('button', { name: 'Move thumbnails' }).click();
+    await expect(page.locator('#workspace-main h1')).toHaveText('Move thumbnails');
+    await expect(page.locator('#workspace-main')).toContainText('Thumbnails moved to');
+}
+
+test('the row moves thumbnails into one folder and back', async ({ page, request }) => {
+    await openSources(page);
+    const row = block(page);
+    await expect(row.getByRole('radio', { name: /Inside each source/ })).toBeChecked();
+    await expect(row.locator('#location-status')).toHaveText('Now: inside each source.');
+    await expect(row.getByRole('button', { name: 'Review move…' })).toHaveAttribute('aria-disabled', 'true');
+    await expect(row.locator('#location-path')).toBeDisabled();
+
+    await row.getByRole('radio', { name: /One folder/ }).check();
+    await expect(row.locator('#location-path')).toBeEnabled();
+    await expect(row.locator('#location-path')).toHaveValue(/default-thumbs$/);
+    await expect(row.locator('#location-status')).toHaveText(/^Now: inside each source\. The suggested folder is next to your library/);
+    await expect(row.getByRole('button', { name: 'Review move…' })).toHaveAttribute('aria-disabled', 'false');
+    await moveFromRow(page);
+    await expect(page.locator('#job-phase')).toHaveText(/^Thumbnails are now read from the new location/);
+    const moved = await files(request);
+    expect(moved.folder.length).toBeGreaterThan(0);
+    expect(moved.folder.every(name => /^[a-z0-9_-]+-[0-9a-f]{8}\//.test(name))).toBe(true);
+    expect(moved.source_thumbnails).toBe(false);
+
+    // Images shows thumbnails served from the folder.
+    const served = page.waitForResponse(response => response.url().includes('/images/serve/thumbnails/')
+        && response.status() === 200);
+    await page.goto(`${workspace}#/images`);
+    await served;
+
+    await openSources(page);
+    await expect(block(page).locator('#location-status')).toHaveText(/^Now: one folder, .*default-thumbs\. Using the default folder/);
+    await block(page).getByRole('radio', { name: /Inside each source/ }).check();
+    await moveFromRow(page);
+    const back = await files(request);
+    expect(back.source_thumbnails).toBe(true);
+    expect(back.folder_exists).toBe(false);
+});
+
+test('a folder the server refuses is explained under the field', async ({ page }) => {
+    await openSources(page);
+    const row = block(page);
+    await row.getByRole('radio', { name: /One folder/ }).check();
+    await row.locator('#location-path').fill('thumbnails');
+    await row.getByRole('button', { name: 'Review move…' }).click();
+    await expect(row.locator('#location-problem')).toHaveText(
+        'Enter a full folder path, such as D:\\Thumbnails or /mnt/fast/thumbnails.');
+    await expect(row.locator('#location-path')).toHaveAttribute('aria-invalid', 'true');
+    await expect(row.locator('#location-path')).toHaveAttribute('aria-describedby', 'location-problem');
+    await expect(dialog(page)).toHaveCount(0);
+    await row.locator('#location-path').fill('/somewhere/else');
+    await expect(row.locator('#location-problem')).toHaveText('');
+});
+
+test('an unplugged folder shows the notice and Recreate folder brings it back', async ({ page, request }) => {
+    await expect(await request.post('/__test__/thumbnail-location', { data: { mode: 'folder' } })).toBeOK();
+    const removed = await request.post('/__test__/thumbnail-folder/remove');
+    expect((await removed.json()).available).toBe(false);
+    await openSources(page);
+    const notice = page.locator('#unavailable-notice');
+    await expect(notice.locator('h2')).toHaveText('The thumbnail folder is unavailable');
+    await expect(notice).toContainText('is missing. Originals are shown and no thumbnails are written until it is back.');
+    await expect(block(page).locator('#location-status')).toHaveText(/· unavailable\. Originals are shown\.$/);
+    await notice.getByRole('button', { name: 'Recreate folder' }).click();
+    await expect(dialog(page).locator('h2')).toHaveText('Recreate the thumbnail folder?');
+    const recreated = page.waitForResponse(response => response.url().endsWith('/location/recreate'));
+    await dialog(page).getByRole('button', { name: 'Recreate folder' }).click();
+    expect((await recreated).status()).toBe(200);
+    await expect(notice).toHaveCount(0);
+    await expect(block(page).locator('#location-status')).toHaveText(/^Now: one folder, .*thumbs\.$/);
+});
+
+test('a folder of your own needs the confirmation before it moves', async ({ page, request }) => {
+    const custom = (await files(request)).custom_path;
+    await openSources(page);
+    const row = block(page);
+    await row.getByRole('radio', { name: /One folder/ }).check();
+    await row.locator('#location-path').fill(custom);
+    await row.getByRole('button', { name: 'Review move…' }).click();
+    const review = dialog(page);
+    await expect(review.locator('dt')).toContainText(['From', 'To', 'Thumbnails', 'Free space', 'Removed after the move']);
+    await expect(review.locator('dd').nth(1)).toHaveText(`${custom} (new folder)`);
+    const move = review.getByRole('button', { name: 'Move thumbnails' });
+    await expect(move).toHaveAttribute('aria-disabled', 'true');
+    await review.getByRole('checkbox', { name: /may create its own folders/ }).check();
+    await expect(move).toHaveAttribute('aria-disabled', 'false');
+    await move.click();
+    await expect(page.locator('#workspace-main')).toContainText('Thumbnails moved to');
+    expect((await files(request)).folder.length).toBeGreaterThan(0);
+});
+
+test('another folder’s marker offers Use this folder', async ({ page, request }) => {
+    await expect(await request.post('/__test__/thumbnail-location', { data: { mode: 'folder' } })).toBeOK();
+    await expect(await request.post('/__test__/thumbnail-folder/foreign')).toBeOK();
+    await openSources(page);
+    const notice = page.locator('#unavailable-notice');
+    await expect(notice).toContainText('holds another thumbnail folder');
+    await notice.getByRole('button', { name: 'Use this folder' }).click();
+    await expect(notice).toHaveCount(0);
+    await expect(block(page).locator('#location-status')).toHaveText(/^Now: one folder, .*thumbs\.$/);
+});
+
+test('Overview shows the notice and Change location opens the row', async ({ page, request }) => {
+    await expect(await request.post('/__test__/thumbnail-location', { data: { mode: 'folder' } })).toBeOK();
+    const removed = await request.post('/__test__/thumbnail-folder/remove');
+    expect((await removed.json()).available).toBe(false);
+    await page.goto(`${workspace}#/system/overview`);
+    const notice = page.locator('#unavailable-notice');
+    await expect(notice.locator('h2')).toHaveText('The thumbnail folder is unavailable');
+    await notice.getByRole('button', { name: 'Change location' }).click();
+    await expect(page).toHaveURL(/#\/system\/sources$/);
+    const sourcesNotice = page.locator('#unavailable-notice');
+    await sourcesNotice.getByRole('button', { name: 'Change location' }).click();
+    await expect(block(page).getByRole('radio', { name: /One folder/ })).toBeChecked();
+    await expect(block(page).locator('#location-path')).toBeFocused();
+});
+
+test('Overview’s Use this folder takes the folder and the notice goes', async ({ page, request }) => {
+    await expect(await request.post('/__test__/thumbnail-location', { data: { mode: 'folder' } })).toBeOK();
+    await expect(await request.post('/__test__/thumbnail-folder/foreign')).toBeOK();
+    await page.goto(`${workspace}#/system/overview`);
+    const notice = page.locator('#unavailable-notice');
+    await expect(notice).toContainText('holds another thumbnail folder');
+    const adopted = page.waitForResponse(response => response.url().endsWith('/location/adopt'));
+    await notice.getByRole('button', { name: 'Use this folder' }).click();
+    expect((await adopted).status()).toBe(200);
+    await expect(notice).toHaveCount(0);
+});
+
+test('Thumbnail maintenance removes the thumbnails of removed sources in folder mode', async ({ page, request }) => {
+    await expect(await request.post('/__test__/thumbnail-location', { data: { mode: 'folder' } })).toBeOK();
+    await expect(await request.post('/__test__/thumbnail-folder/unlisted')).toBeOK();
+    await openSources(page);
+    const generation = page.locator('#workspace-main .operation').filter({ hasText: 'Thumbnail generation' });
+    await generation.getByRole('button', { name: 'Review options' }).click();
+    const scope = dialog(page).locator('#thumb-mode');
+    await expect(scope.locator('option')).toHaveText(['Missing thumbnails only', 'Rebuild all (explicit opt-in)',
+        'Convert outdated (removes older formats)', 'Remove thumbnails of removed sources']);
+    await scope.selectOption('prune');
+    await expect(dialog(page).locator('[role=status]')).toHaveText(
+        /^1 source no longer listed: 1 thumbnail \(.+\) are removed\.$/);
+    await expect(dialog(page).locator('dl.meta-grid dt')).toHaveText(['gone-0123abcd']);
+    await expect(dialog(page)).toContainText('If a source comes back, its thumbnails are made again as you browse.');
+    await scope.selectOption('missing');
+    await expect(dialog(page).locator('dl.meta-grid')).toBeHidden();
+    await scope.selectOption('prune');
+    await dialog(page).getByRole('button', { name: 'Start' }).click();
+    await expect(page.locator('#workspace-main h1')).toHaveText('Remove thumbnails of removed sources');
+    await expect(page.locator('#workspace-main')).toContainText('1 thumbnails removed');
+});
+
+test('a disk without the space shows the review but cannot move', async ({ page, request }) => {
+    await expect(await request.post('/__test__/thumbnail-space/full')).toBeOK();
+    await openSources(page);
+    const row = block(page);
+    await row.getByRole('radio', { name: /One folder/ }).check();
+    const planned = page.waitForResponse(response => response.url().endsWith('/images/thumbnails/plan'));
+    await row.getByRole('button', { name: 'Review move…' }).click();
+    expect((await planned).status()).toBe(507);
+    const review = dialog(page);
+    await expect(review.locator('.review-refusal')).toHaveText(/^There is not enough space for the thumbnails: 1 B free/);
+    await expect(review.getByRole('button', { name: 'Move thumbnails' })).toHaveAttribute('aria-disabled', 'true');
+    await review.getByRole('button', { name: 'Cancel' }).click();
+    await expect(dialog(page)).toHaveCount(0);
+});

tests/fixtures/legacy_contract_snapshot.json

@@ -125,6 +125,9 @@
     "POST /prompt_manager/images/removed/return",
     "POST /prompt_manager/images/rescope",
     "POST /prompt_manager/images/reveal",
+    "POST /prompt_manager/images/thumbnails/location/adopt",
+    "POST /prompt_manager/images/thumbnails/location/recreate",
+    "POST /prompt_manager/images/thumbnails/plan",
     "POST /prompt_manager/jobs/autotag",
     "POST /prompt_manager/jobs/duplicates",
     "POST /prompt_manager/jobs/image-tags",
@@ -215,6 +218,7 @@
     "thumbnail_encoders.webp",
     "thumbnail_location",
     "thumbnail_location.available",
+    "thumbnail_location.cleanup_pending",
     "thumbnail_location.default_path",
     "thumbnail_location.mode",
     "thumbnail_location.path",

tests/js/api-client.test.js

@@ -371,6 +371,16 @@ test("ApiError keeps the job that applied a used plan (plan-used, System S7)", (
     }
 });
 
+test("ApiError keeps a move refused for space with its review (#320)", () => {
+    const plan = { totals: { files: 3 }, insufficient: { text: "Not enough space" } };
+    assert.equal(new ApiError(507, "no space", "insufficient-space", null, { plan }).plan, plan);
+    for (const bad of [null, "x", [1], undefined]) {
+        assert.equal(new ApiError(507, "no space", "insufficient-space", null, { plan: bad }).plan, null);
+    }
+    assert.equal(new ApiError(409, "busy", "busy", null, { plan }).plan, null);
+    assert.equal(new ApiError(507, "no space", "insufficient-space").plan, null);
+});
+
 test("ApiError keeps a validated running job on busy and a validated source on invalid-source", () => {
     const body = { running: { id: "a".repeat(32), title: "Scan for metadata" }, source: "output/gone" };
     const busy = new ApiError(409, "busy", "busy", null, body);

tests/js/workspace-system-thumbnail-location.test.js

@@ -0,0 +1,170 @@
+// Thumbnail location (#320, D196–D201): the row's state and status line, the move review's rows, the unavailable
+// notice's text, the Activity views of a move and a prune, the prune scope's plan line and the Move library row.
+const test = require('node:test');
+const assert = require('node:assert/strict');
+
+const locationModule = import('../../web/js/workspace/system-thumbnail-location.mjs');
+const kindsModule = import('../../web/js/workspace/activity-kinds.mjs');
+const reviewModule = import('../../web/js/workspace/thumbnail-review.mjs');
+const libraryModule = import('../../web/js/workspace/library-location.mjs');
+
+const SOURCE = Object.freeze({ mode: 'source', path: null, default_path: '/data/thumbnails', available: true,
+    problem: null, cleanup_pending: 0 });
+const FOLDER = Object.freeze({ mode: 'folder', path: 'E:\\Thumbnails', default_path: '/data/thumbnails',
+    available: true, problem: null, cleanup_pending: 0 });
+
+const PLAN = Object.freeze({
+    from: { mode: 'source', path: null, available: true },
+    to: { mode: 'folder', path: 'E:\\Thumbnails', exists: true, will_create: false, default: false },
+    totals: { sources: 3, files: 8512, bytes: 1.2 * 1024 ** 3, orphans: 41, temps: 2, links: 1, left_in_place: 1,
+        too_long: 32, unlisted_sources: 0, unlisted_files: 0 },
+    too_long_listed: ['Archive 2025 · a_thumb.webp', 'Archive 2025 · b_thumb.webp'],
+    left_listed: ['Archive 2025 · notes.txt'],
+    unlisted: [],
+    free_bytes: 1.6 * 1024 ** 4,
+    confirm_required: true,
+    token: 't',
+});
+
+test('the status line says where previews are now', async () => {
+    const { statusText } = await locationModule;
+    assert.equal(statusText(SOURCE), 'Now: inside each source.');
+    assert.equal(statusText(FOLDER), 'Now: one folder, E:\\Thumbnails.');
+    assert.equal(statusText({ ...FOLDER, path: '/data/thumbnails' }),
+        'Now: one folder, /data/thumbnails. Using the default folder next to your library.');
+    assert.equal(statusText({ ...FOLDER, available: false, problem: 'missing' }),
+        'Now: one folder, E:\\Thumbnails · unavailable. Originals are shown.');
+    assert.match(statusText({ ...SOURCE, cleanup_pending: 1 }), /Old thumbnails in 1 place wait to be removed/);
+    assert.equal(statusText(undefined), '');
+    assert.match(statusText(SOURCE, { mode: 'folder', path: '/data/thumbnails' }),
+        /^Now: inside each source\. The suggested folder is next to your library/);
+    assert.equal(statusText(SOURCE, { mode: 'folder', path: '/other' }), 'Now: inside each source.');
+});
+
+test('Review move… is offered only for another location', async () => {
+    const { choiceChanged } = await locationModule;
+    assert.equal(choiceChanged(SOURCE, { mode: 'source' }), false);
+    assert.equal(choiceChanged(SOURCE, { mode: 'folder', path: ' ' }), false);
+    assert.equal(choiceChanged(SOURCE, { mode: 'folder', path: '/data/thumbnails' }), true);
+    assert.equal(choiceChanged(FOLDER, { mode: 'folder', path: 'E:\\Thumbnails' }), false);
+    assert.equal(choiceChanged(FOLDER, { mode: 'folder', path: 'F:\\Thumbs' }), true);
+    assert.equal(choiceChanged(FOLDER, { mode: 'source' }), true);
+    assert.equal(choiceChanged({ ...SOURCE, problem: 'invalid' }, { mode: 'source' }), true);
+    assert.equal(choiceChanged(undefined, { mode: 'source' }), false);
+});
+
+test('the review lists what moves, what is removed and what stays', async () => {
+    const { reviewRows } = await locationModule;
+    const rows = Object.fromEntries(reviewRows(PLAN));
+    assert.equal(rows.From, 'Inside each source (3 sources)');
+    assert.equal(rows.To, 'E:\\Thumbnails (empty folder)');
+    assert.equal(rows.Thumbnails, '8,512 in 3 sources · 1.2 GB');
+    assert.equal(rows['Free space'], '1638.4 GB free');
+    assert.match(rows['Too long for Windows'][0], /^32 thumbnails cannot be stored there/);
+    assert.deepEqual(rows['Too long for Windows'][1], [...PLAN.too_long_listed, 'and 30 more']);
+    assert.deepEqual(rows['Removed after the move'][1],
+        ['41 thumbnails whose original is gone', '2 unfinished temporary files']);
+    assert.equal(rows['Left in place'][0], '1 file that is not a thumbnail:');
+    assert.equal(rows['Links skipped'], '1 link inside a thumbnails folder is not followed');
+});
+
+test('the review of a move back names the folder and the removed sources', async () => {
+    const { reviewRows } = await locationModule;
+    const back = reviewRows({ ...PLAN, from: { mode: 'folder', path: '/data/thumbnails' },
+        to: { mode: 'source', path: null }, unlisted: [{ key: 'old-1c22e0a4', files: 240 }],
+        totals: { sources: 1, files: 1, bytes: 10 }, free_bytes: undefined });
+    const rows = Object.fromEntries(back);
+    assert.equal(rows.From, '/data/thumbnails');
+    assert.equal(rows.To, 'Inside each source (1 source)');
+    assert.equal(rows['Free space'], undefined);
+    assert.match(rows['Removed after the move'][0], /The folder itself is removed once empty/);
+    assert.deepEqual(rows['Removed after the move'][1],
+        ['1 source no longer listed: 240 thumbnails are removed; they are made again if it returns']);
+    // Nothing else removed: one full stop, never "created it.." (live test of #592).
+    const plain = Object.fromEntries(reviewRows({ ...PLAN, from: { mode: 'folder', path: '/data/thumbnails' },
+        to: { mode: 'source', path: null }, unlisted: [], totals: { sources: 1, files: 1, bytes: 10 } }));
+    assert.match(plain['Removed after the move'][0], /Prompt Manager created it\.$/);
+    const fresh = Object.fromEntries(reviewRows({ ...PLAN, to: { mode: 'folder', path: '/d', default: true, will_create: true } }));
+    assert.equal(fresh.To, '/d (new, next to your library)');
+});
+
+test('the notice says why the folder is unavailable', async () => {
+    const { noticeText } = await locationModule;
+    assert.match(noticeText({ ...FOLDER, problem: 'missing' }), /^E:\\Thumbnails is missing\. Originals are shown/);
+    assert.match(noticeText({ ...FOLDER, problem: 'unreachable' }), /can’t be reached/);
+    assert.match(noticeText({ ...FOLDER, problem: 'marker-mismatch' }), /holds another thumbnail folder/);
+    assert.match(noticeText({ ...FOLDER, problem: 'invalid' }), /config\.json can’t be used/);
+    assert.match(noticeText({ ...FOLDER, problem: null }), /is being checked/);
+});
+
+test('sizes are written as the server writes them', async () => {
+    const { sizeText } = await locationModule;
+    assert.equal(sizeText(512), '512 B');
+    assert.equal(sizeText(85.2 * 1024 * 1024), '85.2 MB');
+});
+
+test('a move and a prune show their own tiles in Activity', async () => {
+    const { viewOf, movePhase, finishToast } = await kindsModule;
+    const move = { kind: 'thumbnails', state: 'running', cancellable: true, scope: { mode: 'move', from: 'source' },
+        counters: { moved: 3, skipped: 0, left: 1, errors: 2 }, status_line: '' };
+    assert.deepEqual(viewOf(move).tiles.map(([group, label]) => label), ['Moved', 'Already there', 'Left in place', 'Errors']);
+    assert.equal(viewOf(move).cancelLabel, 'Cancel move');
+    assert.match(movePhase(move), /^Step 1 of 3: copying\. Thumbnails are still read from inside each source/);
+    assert.match(movePhase({ ...move, cancellable: false }), /can no longer be cancelled/);
+    assert.match(movePhase({ ...move, scope: { mode: 'move', phase: 'remove' } }), /Removing the old copies/);
+    assert.match(movePhase({ ...move, state: 'finished' }), /1 file that is not a thumbnail was left in place/);
+    assert.match(movePhase({ ...move, state: 'cancelled' }), /^Nothing changed/);
+    assert.match(movePhase({ ...move, state: 'finished', counters: { moved: 3, left: 2 } }),
+        /2 files that are not thumbnails were left in place\.$/);
+    assert.equal(movePhase({ ...move, state: 'finished', counters: undefined }), 'Thumbnails are now read from the new location.');
+    assert.match(movePhase({ ...move, scope: { mode: 'move', from: 'folder' } }), /read from the old folder until the switch/);
+    const done = { ...move, state: 'finished', status_line: 'Thumbnails moved to E:\\T: 3 moved · old copies removed.' };
+    assert.equal(finishToast(done), done.status_line);
+    const prune = { kind: 'thumbnails', state: 'running', cancellable: true, scope: { mode: 'prune' },
+        counters: { removed: 1, left: 0, errors: 0 } };
+    assert.deepEqual(viewOf(prune).tiles.map(([, label]) => label), ['Removed', 'Left in place', 'Errors']);
+    assert.equal(viewOf(move).line(move), '');
+    assert.equal(viewOf(done).line(done), done.status_line);
+    assert.equal(viewOf(move).kept(move), '3 moved');
+    assert.equal(finishToast({ ...move, state: 'cancelled', status_line: '' }), viewOf(move).finishToast({ ...move,
+        state: 'cancelled', status_line: '' }));
+    const pruned = { ...prune, state: 'finished', status_line: '1 thumbnails removed' };
+    assert.equal(viewOf(prune).line(prune), '');
+    assert.equal(viewOf(pruned).line(pruned), '1 thumbnails removed');
+    assert.equal(viewOf(pruned).line({ ...pruned, status_line: null }), '');
+    assert.equal(viewOf(done).line({ ...done, status_line: null }), '');
+    assert.equal(finishToast(pruned), '1 thumbnails removed');
+    assert.equal(viewOf(prune).kept(prune), '1 removed');
+    assert.match(finishToast({ ...prune, state: 'finished', status_line: '', title: 'Remove thumbnails of removed sources' }),
+        /Remove thumbnails of removed sources/);
+    const generate = { kind: 'thumbnails', state: 'running', cancellable: true, scope: { mode: 'missing' },
+        counters: { generated: 0, skipped: 0, errors: 0 } };
+    assert.deepEqual(viewOf(generate).tiles.map(([, label]) => label), ['Generated', 'Skipped', 'Errors']);
+});
+
+test('the prune scope reads its plan line from the server', async () => {
+    const { planText, PLAN_FAILED, PRUNE_OPTION } = await reviewModule;
+    assert.equal(PRUNE_OPTION[1], 'Remove thumbnails of removed sources');
+    assert.equal(planText({ mode: 'prune', line: '2 sources no longer listed: 1,240 thumbnails (85.2 MB) are removed.' }),
+        '2 sources no longer listed: 1,240 thumbnails (85.2 MB) are removed.');
+    assert.equal(planText({ mode: 'prune' }), PLAN_FAILED);
+});
+
+test('the prune review names the folder and each removed source', async () => {
+    const { pruneText, pruneEntries } = await reviewModule;
+    assert.match(pruneText({ folder: 'E:\\Thumbs' }), /^Only Prompt Manager’s thumbnails in their folders inside E:\\Thumbs are /);
+    assert.match(pruneText({}), /inside the thumbnail folder are removed\. If a source comes back/);
+    assert.deepEqual(pruneEntries({ sources: [{ key: 'gone-0123abcd', files: 1240, bytes: 85.2 * 1024 * 1024 },
+        { key: 'one-0123abcd', files: 1, bytes: 512 }] }),
+    [['gone-0123abcd', '1,240 thumbnails · 85.2 MB'], ['one-0123abcd', '1 thumbnail · 512 B']]);
+    assert.deepEqual(pruneEntries(null), []);
+    assert.deepEqual(pruneEntries({ sources: 'x' }), []);
+});
+
+test('Move library says where thumbnails kept in one folder stay', async () => {
+    const { reviewGrid } = await libraryModule;
+    const base = { path: 'p', target: 't', size_bytes: 0, config_moves: true };
+    assert.equal(reviewGrid(base).some(([term]) => term === 'Thumbnails'), false);
+    const rows = Object.fromEntries(reviewGrid({ ...base, thumbnails_folder: '/data/thumbnails' }));
+    assert.equal(rows.Thumbnails, 'Stay in /data/thumbnails. Move them in System › Sources & scans.');
+});

tests/test_api_library_location.py

@@ -91,6 +91,8 @@ class TestLocationRoute(LibraryAPITestCase):
                 "size_bytes": os.path.getsize(self.api.db.model.db_path),
                 "target_source": "default",
                 "config_moves": True,
+                # #320 D201: previews kept inside each source (no folder).
+                "thumbnails_folder": None,
             },
         )
 

tests/test_api_thumbnail_location.py

@@ -189,6 +189,62 @@ class TestMarker(unittest.TestCase):
         # No temp file is left beside it.
         self.assertEqual(os.listdir(self.folder), [pl.MARKER_NAME])
 
+    def test_two_processes_adding_keys_at_once_keep_every_key(self):
+        """Installs sharing one folder (bot review, P2): the read-modify-write
+        of the marker is locked across processes, so no writer drops another's
+        key. Each child widens the window by pausing inside ``mutate``."""
+        import subprocess
+
+        pl.init_folder(self.folder)
+        child = (
+            "import sys, time\n"
+            "sys.path.insert(0, sys.argv[1])\n"
+            "from py.api import preview_location as pl\n"
+            "folder, tag, start = sys.argv[2], sys.argv[3], float(sys.argv[4])\n"
+            "while time.time() < start: time.sleep(0.001)\n"
+            "for n in range(8):\n"
+            "    def add(m, n=n):\n"
+            "        time.sleep(0.01)\n"
+            "        m['keys'][f'{tag}-{n:08x}'] = {'install': tag}\n"
+            "    pl.write_marker(folder, add)\n"
+        )
+        import time
+
+        start = str(time.time() + 1.0)
+        procs = [
+            subprocess.Popen(  # nosec B603 - this interpreter, fixed arguments
+                [sys.executable, "-c", child, str(REPO), self.folder, tag, start]
+            )
+            for tag in ("a", "b", "c")
+        ]
+        for proc in procs:
+            self.assertEqual(proc.wait(timeout=60), 0)
+        keys = pl.read_marker(self.folder)["keys"]
+        self.assertEqual(len(keys), 24, sorted(keys))
+        self.assertEqual(os.listdir(self.folder), [pl.MARKER_NAME])
+
+    def test_a_lock_left_by_a_crashed_writer_is_taken_over_when_stale(self):
+        pl.init_folder(self.folder)
+        lock = os.path.join(self.folder, pl.MARKER_LOCK_NAME)
+        with open(lock, "w", encoding="utf-8") as handle:
+            handle.write("crashed")
+        old = pl.time.time() - pl.MARKER_LOCK_STALE - 5
+        os.utime(lock, (old, old))
+        pl.write_marker(self.folder, lambda m: m["keys"].update({"a-0123abcd": {}}))
+        self.assertIn("a-0123abcd", pl.read_marker(self.folder)["keys"])
+        self.assertFalse(os.path.exists(lock))
+
+    def test_a_held_lock_times_out_without_writing(self):
+        pl.init_folder(self.folder)
+        lock = os.path.join(self.folder, pl.MARKER_LOCK_NAME)
+        with open(lock, "w", encoding="utf-8") as handle:
+            handle.write("held")
+        with patch.object(pl, "MARKER_LOCK_TIMEOUT", 0.05):
+            with self.assertRaises(TimeoutError):
+                pl.write_marker(self.folder, lambda m: m["keys"].update({"x": {}}))
+        self.assertTrue(os.path.exists(lock))  # never taken from its holder
+        self.assertEqual(pl.read_marker(self.folder)["keys"], {})
+
     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")
@@ -263,6 +319,22 @@ class TestValidation(unittest.TestCase):
         os.mkdir(folder)
         self.assertEqual(self.problem(folder, current_path=folder), "location-same")
 
+    def test_a_folder_inside_the_folder_in_use_is_refused(self):
+        # Codex P2 on #592: the old key folders are removed after the move,
+        # and with them a target nested inside one of them.
+        folder = os.path.join(self.base, "thumbs")
+        os.mkdir(folder)
+        pl.init_folder(folder)
+        for nested in (
+            os.path.join(folder, "output-0123abcd", "nested"),
+            os.path.join(folder, "beside"),
+        ):
+            with self.subTest(nested=nested):
+                self.assertEqual(
+                    self.problem(nested, current_path=folder),
+                    "location-inside-current",
+                )
+
     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)
@@ -302,7 +374,7 @@ class TestValidation(unittest.TestCase):
     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.assertIn(status, (400, 409, 507))
                 self.assertTrue(text.endswith("."))
 
 

tests/test_preview_probe.py

@@ -0,0 +1,93 @@
+"""The thumbnail folder check answers for the folder as it is now (#320).
+
+A check that started before the caller asked can miss what just happened: #592
+hit it in the merge queue, where a folder deleted right after
+``configure`` woke the probe stayed "available" until the next probe.
+"""
+
+import sys
+import threading
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+
+from py.api import preview_location, preview_store  # noqa: E402
+
+WAIT = 5
+HERE = (True, None, {"id": "x"})
+GONE = (False, "missing", None)
+KEY = ("/thumbs", "x")
+
+
+class TestProbeCheck(unittest.TestCase):
+    def setUp(self):
+        self.probe = preview_store._Probe()
+
+    def test_a_finished_check_nobody_collected_is_not_the_answer(self):
+        with patch.object(preview_location, "folder_state", return_value=HERE):
+            with self.probe._lock:
+                _key, done, _result = self.probe._start_check(*KEY)
+            self.assertTrue(done.wait(WAIT))
+        with patch.object(preview_location, "folder_state", return_value=GONE):
+            self.assertEqual(self.probe.check(*KEY), GONE)
+
+    def test_a_check_already_running_is_followed_by_a_new_one(self):
+        started = threading.Event()
+        release = threading.Event()
+        calls = []
+
+        def state(_path, _expected_id):
+            calls.append(threading.current_thread().name)
+            if len(calls) == 1:
+                started.set()
+                self.assertTrue(release.wait(WAIT))
+                return HERE
+            return GONE
+
+        answers = []
+        with patch.object(preview_location, "folder_state", state):
+            with self.probe._lock:
+                self.probe._start_check(*KEY)
+            self.assertTrue(started.wait(WAIT))
+            caller = threading.Thread(
+                target=lambda: answers.append(self.probe.check(*KEY))
+            )
+            caller.start()
+            release.set()
+            caller.join(WAIT)
+        self.assertFalse(caller.is_alive())
+        self.assertEqual(answers, [GONE])
+        self.assertEqual(len(calls), 2)  # one after the other, never two at once
+
+    def test_a_stuck_check_is_never_doubled(self):
+        release = threading.Event()
+        self.addCleanup(release.set)
+        calls = []
+
+        def state(_path, _expected_id):
+            calls.append(1)
+            release.wait(WAIT)
+            return HERE
+
+        with (
+            patch.object(preview_location, "folder_state", state),
+            patch.object(preview_store, "IO_BUDGET", 0.05),
+        ):
+            with self.probe._lock:
+                self.probe._start_check(*KEY)
+            with self.assertRaises(TimeoutError):
+                self.probe.check(*KEY)
+        self.assertEqual(len(calls), 1)
+
+    def test_an_error_from_the_new_check_reaches_the_caller(self):
+        with patch.object(
+            preview_location, "folder_state", side_effect=OSError("share gone")
+        ):
+            with self.assertRaises(OSError):
+                self.probe.check(*KEY)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_preview_store_folder.py

@@ -469,6 +469,71 @@ class TestHostileMarker(FolderModeCase):
         self.assertEqual(victim.read_bytes(), b"keep me")
         self.assertEqual(sorted(p.name for p in outside.iterdir()), ["v_thumb.webp"])
 
+    def link_key_folder_outside(self):
+        """The source's key folder replaced by a link to a folder outside."""
+        outside = self.comfy_dir / "outside"
+        outside.mkdir()
+        store = self.key_folder()
+        store.rmdir()
+        try:
+            store.symlink_to(outside, target_is_directory=True)
+        except OSError:
+            self.skipTest("symlinks not permitted")
+        return outside
+
+    @unittest.skipUnless(hasattr(os, "symlink"), "needs symlinks")
+    async def test_convert_never_plans_or_deletes_through_a_linked_key_folder(self):
+        outside = self.link_key_folder_outside()
+        old = outside / "c_thumb.png"
+        old.write_bytes(b"keep me")
+        make_png(self.output_dir / "c.png", size=(640, 480))
+        # A linked store is refused before any planning (bot review, P1).
+        self.assertIsNone(preview_store.ensure_store(self.output_dir))
+        plan = self.api._plan_gallery_thumbnails([self.output_dir], mode="convert")
+        self.assertEqual(plan["targets"], [])
+        # And every delete passes the ownership check at the moment it runs.
+        removed, kept = self.api._remove_old_previews([(old, 7)])
+        self.assertEqual((removed, kept), (0, [old]))
+        self.assertEqual(old.read_bytes(), b"keep me")
+
+    @unittest.skipUnless(hasattr(os, "symlink"), "needs symlinks")
+    async def test_the_temp_sweep_never_walks_into_a_link(self):
+        from py.api import thumbnail_encode
+
+        outside = self.comfy_dir / "outside"
+        outside.mkdir()
+        victim = outside / ".v_thumb.part-1-1.webp"
+        victim.write_bytes(b"keep me")
+        os.utime(victim, (1, 1))
+        try:
+            (self.key_folder() / "sub").symlink_to(outside, target_is_directory=True)
+        except OSError:
+            self.skipTest("symlinks not permitted")
+        real_walk = os.walk
+
+        def windows_walk(top, *args, **kwargs):
+            # Windows walks into a junction even with followlinks=False.
+            kwargs["followlinks"] = True
+            return real_walk(top, *args, **kwargs)
+
+        with patch.object(thumbnail_encode.os, "walk", windows_walk):
+            self.assertEqual(self.api.sweep_thumbnail_temps(), 0)
+            thumbnail_encode.sweep_stray_temps(self.key_folder(), min_age=0)
+            resp = await self.client.post("/prompt_manager/images/clear-thumbnails")
+        self.assertEqual(resp.status, 200)
+        self.assertEqual(victim.read_bytes(), b"keep me")
+
+    def test_the_temp_sweep_leaves_a_temp_the_store_does_not_own(self):
+        from py.api import thumbnail_encode
+
+        temp = self.key_folder() / ".w_thumb.part-1-1.webp"
+        temp.write_bytes(b"x")
+        with patch.object(preview_store, "may_write", return_value=False):
+            self.assertEqual(
+                thumbnail_encode.sweep_stray_temps(self.key_folder(), min_age=0), (0, 0)
+            )
+        self.assertTrue(temp.exists())
+
     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")

tests/test_thumbnail_move.py

@@ -0,0 +1,542 @@
+"""Move thumbnails (#320 PR 3): the reviewed job in every direction, its
+failure, cancel and restart paths, Recreate and the removed-sources scope.
+
+Each test runs against a temp config.json (``AdminAPITestCase``) and a temp
+journal; jobs are awaited through their task, never a timer.
+"""
+
+import json
+import os
+import re
+import sys
+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_api_admin import AdminAPITestCase
+    from tests.test_api_images import make_png
+except ImportError:  # discovered with tests/ as the top-level directory
+    from test_api_admin import AdminAPITestCase
+    from test_api_images import make_png
+
+from py.api import preview_location, preview_store  # noqa: E402
+from py.api import thumbnail_move_files as files  # noqa: E402
+from py.api import thumbnail_move_plan as move_plan  # noqa: E402
+from py.config import PromptManagerConfig  # noqa: E402
+
+REPO = Path(__file__).resolve().parent.parent
+PLAN = "/prompt_manager/images/thumbnails/plan"
+START = "/prompt_manager/jobs/thumbnails"
+
+
+class MoveCase(AdminAPITestCase):
+    async def asyncSetUp(self):
+        await super().asyncSetUp()
+        self.api._get_all_output_dirs = lambda: [self.output_dir]
+        self.api.recover_thumbnail_move()
+        self.journal = Path(self.tmpdir) / "data" / "thumbnail-move.json"
+        for target in (move_plan,):
+            patcher = patch.object(target, "journal_path", lambda: str(self.journal))
+            patcher.start()
+            self.addCleanup(patcher.stop)
+        default = patch(
+            "py.api.thumbnail_move_plan.default_path",
+            lambda: str(Path(self.tmpdir) / "default-thumbs"),
+        )
+        default.start()
+        self.addCleanup(default.stop)
+        self.addCleanup(preview_store.configure, None)
+        self.addCleanup(setattr, PromptManagerConfig, "THUMBNAIL_LOCATION", None)
+        self.old_store = self.output_dir / "thumbnails"
+        self.folder = Path(self.tmpdir) / "thumbs"
+        # Two media files with previews, an orphan, a temp and a note.
+        make_png(self.output_dir / "a.png", size=(8, 8))
+        make_png(self.output_dir / "sub" / "b.png", size=(8, 8))
+        make_png(self.old_store / "a_thumb.webp", size=(4, 4))
+        make_png(self.old_store / "sub" / "b_thumb.webp", size=(4, 4))
+        make_png(self.old_store / "gone_thumb.webp", size=(4, 4))
+        (self.old_store / ".a_thumb.part-1-1.webp").write_bytes(b"t")
+        (self.old_store / "notes.txt").write_text("mine", encoding="utf-8")
+
+    async def plan(self, location):
+        resp = await self.client.post(PLAN, json={"mode": "move", "location": location})
+        return resp.status, await resp.json()
+
+    async def move(self, location, confirmed=True):
+        status, plan = await self.plan(location)
+        self.assertEqual(status, 200, plan)
+        resp = await self.client.post(
+            START, json={"mode": "move", "token": plan["token"], "confirmed": confirmed}
+        )
+        body = await resp.json()
+        self.assertEqual(resp.status, 202, body)
+        job = self.api._jobs.get(body["job"]["id"])
+        await job.task
+        return plan, job
+
+    def config(self):
+        with open(self.config_path, encoding="utf-8") as handle:
+            return json.load(handle)
+
+    def key_files(self, folder=None):
+        folder = folder or self.folder
+        return sorted(
+            p.relative_to(folder).as_posix()
+            for p in folder.rglob("*")
+            if p.is_file() and p.name != preview_location.MARKER_NAME
+        )
+
+
+class TestMoveToAFolder(MoveCase):
+    async def test_review_counts_what_moves_is_removed_and_stays(self):
+        status, plan = await self.plan({"mode": "folder", "path": str(self.folder)})
+        self.assertEqual(status, 200, plan)
+        totals = plan["totals"]
+        self.assertEqual(
+            (
+                totals["files"],
+                totals["orphans"],
+                totals["temps"],
+                totals["left_in_place"],
+            ),
+            (2, 1, 1, 1),
+        )
+        self.assertTrue(plan["confirm_required"])
+        self.assertTrue(plan["to"]["will_create"])
+        self.assertFalse(self.folder.exists())  # the review writes nothing
+
+    async def test_a_move_copies_switches_and_removes_old_copies(self):
+        before = (self.output_dir / "a.png").read_bytes()
+        _plan, job = await self.move({"mode": "folder", "path": str(self.folder)})
+        self.assertEqual(job.state, "finished", job.log)
+        self.assertEqual(job.counters["moved"], 2)
+        keys = self.key_files()
+        self.assertEqual(len(keys), 2)
+        self.assertTrue(all(re.match(r"^output-[0-9a-f]{8}/", k) for k in keys))
+        location = self.config()["thumbnails"]["location"]
+        self.assertEqual(
+            (location["mode"], location["path"]), ("folder", str(self.folder))
+        )
+        self.assertEqual(location["cleanup"], [])
+        self.assertEqual(preview_store.current().mode, "folder")
+        # Old copies, the orphan and the temp went; the note stays, so its
+        # folder stays; originals are untouched.
+        self.assertEqual(
+            sorted(p.name for p in self.old_store.rglob("*")), ["notes.txt"]
+        )
+        self.assertEqual((self.output_dir / "a.png").read_bytes(), before)
+        self.assertFalse(self.journal.exists())
+        copied = next(self.folder.rglob("a_thumb.webp"))
+        self.assertEqual(copied.stat().st_mtime_ns % files.MTIME_STEP_NS, 0)
+
+    async def test_the_move_back_removes_the_folder_it_made(self):
+        await self.move({"mode": "folder", "path": str(self.folder)})
+        _plan, job = await self.move({"mode": "source"})
+        self.assertEqual(job.state, "finished", job.log)
+        self.assertTrue((self.old_store / "a_thumb.webp").is_file())
+        self.assertTrue((self.old_store / "sub" / "b_thumb.webp").is_file())
+        self.assertFalse(self.folder.exists())
+        self.assertEqual(self.config()["thumbnails"]["location"]["mode"], "source")
+
+    async def test_folder_to_folder_keeps_the_key_names(self):
+        await self.move({"mode": "folder", "path": str(self.folder)})
+        first = self.key_files()
+        other = Path(self.tmpdir) / "thumbs2"
+        _plan, job = await self.move({"mode": "folder", "path": str(other)})
+        self.assertEqual(job.state, "finished", job.log)
+        self.assertEqual(self.key_files(other), first)
+        self.assertFalse(self.folder.exists())
+
+    async def test_previews_of_sources_no_longer_listed_go_when_leaving(self):
+        await self.move({"mode": "folder", "path": str(self.folder)})
+        setting = preview_store.setting()
+
+        def add(marker):
+            marker["keys"]["old-0123abcd"] = {
+                "path": "/gone",
+                "identity": "/gone",
+                "install": setting.install,
+            }
+
+        preview_location.write_marker(str(self.folder), add)
+        make_png(self.folder / "old-0123abcd" / "x_thumb.webp", size=(4, 4))
+        preview_store.refresh()
+        status, plan = await self.plan({"mode": "source"})
+        self.assertEqual(status, 200)
+        self.assertEqual(plan["totals"]["unlisted_sources"], 1)
+        _, job = await self.move({"mode": "source"})
+        self.assertEqual(job.state, "finished", job.log)
+        self.assertFalse(self.folder.exists())
+
+    async def test_confirmation_and_a_stale_token(self):
+        status, plan = await self.plan({"mode": "folder", "path": str(self.folder)})
+        resp = await self.client.post(
+            START, json={"mode": "move", "token": plan["token"]}
+        )
+        self.assertEqual((await resp.json())["code"], "confirmation-required")
+        resp = await self.client.post(
+            START, json={"mode": "move", "token": "nope", "confirmed": True}
+        )
+        self.assertEqual(resp.status, 409)
+        self.assertEqual((await resp.json())["code"], "plan-stale")
+
+    async def test_a_folder_that_is_too_small_is_refused(self):
+        with patch.object(move_plan, "_free", lambda _path: 1):
+            status, body = await self.plan({"mode": "folder", "path": str(self.folder)})
+        self.assertEqual(status, 507)
+        self.assertEqual(body["code"], "insufficient-space")
+
+    async def test_validation_refusals_come_back_in_place(self):
+        status, body = await self.plan(
+            {"mode": "folder", "path": str(self.output_dir / "thumbs")}
+        )
+        self.assertEqual((status, body["code"]), (400, "location-served"))
+        status, body = await self.plan({"mode": "source"})
+        self.assertEqual(body["code"], "location-same")
+
+
+class TestFailures(MoveCase):
+    async def test_a_per_file_error_still_switches_and_removes_its_old_copy(self):
+        real = files.copy_one
+
+        def failing(source, destination, *args, **kwargs):
+            if Path(source).name == "a_thumb.webp":
+                raise files.CopyFailed("Could not be read; made again on view")
+            return real(source, destination, *args, **kwargs)
+
+        with patch.object(files, "copy_one", failing):
+            _plan, job = await self.move({"mode": "folder", "path": str(self.folder)})
+        self.assertEqual(job.state, "finished")
+        self.assertEqual((job.counters["moved"], job.counters["errors"]), (1, 1))
+        self.assertFalse((self.old_store / "a_thumb.webp").exists())
+
+    async def test_a_folder_wide_error_changes_nothing(self):
+        def gone(*_args, **_kwargs):
+            raise files.FolderGone("disk full")
+
+        with patch.object(files, "copy_one", gone):
+            _plan, job = await self.move({"mode": "folder", "path": str(self.folder)})
+        self.assertEqual(job.state, "failed")
+        self.assertIn("nothing changed", job.status_line)
+        self.assertFalse(self.folder.exists())
+        self.assertTrue((self.old_store / "a_thumb.webp").is_file())
+        self.assertEqual(preview_store.current().mode, "source")
+        self.assertFalse(self.journal.exists())
+
+    async def test_cancel_before_the_switch_changes_nothing(self):
+        status, plan = await self.plan({"mode": "folder", "path": str(self.folder)})
+        real = self.api._move_copy_one
+
+        def cancel_first(pair, **kwargs):
+            job = self.api._jobs.active("thumbnails")
+            job.cancelled = True
+            return real(pair, **kwargs)
+
+        with patch.object(self.api, "_move_copy_one", cancel_first):
+            resp = await self.client.post(
+                START, json={"mode": "move", "token": plan["token"], "confirmed": True}
+            )
+            job = self.api._jobs.get((await resp.json())["job"]["id"])
+            await job.task
+        self.assertEqual(job.state, "cancelled")
+        self.assertFalse(self.folder.exists())
+        self.assertEqual(preview_store.current().mode, "source")
+
+    async def test_an_exit_during_the_copy_leaves_the_journal_for_the_next_start(self):
+        status, plan = await self.plan({"mode": "folder", "path": str(self.folder)})
+        real = self.api._move_copy_one
+        exiting = {"now": False}
+
+        def exit_after(pair, **kwargs):
+            result = real(pair, **kwargs)
+            exiting["now"] = True  # the interpreter starts exiting
+            return result
+
+        from py.api import thumbnail_move
+
+        with (
+            patch.object(self.api, "_move_copy_one", exit_after),
+            patch.object(
+                thumbnail_move.exit_signal, "requested", lambda: exiting["now"]
+            ),
+        ):
+            resp = await self.client.post(
+                START, json={"mode": "move", "token": plan["token"], "confirmed": True}
+            )
+            job = self.api._jobs.get((await resp.json())["job"]["id"])
+            await job.task
+        self.assertTrue(self.journal.exists())
+        self.assertTrue(self.key_files())  # partial copies were left
+        # The next start undoes it: copies and the created folder go.
+        self.api._move_recovered = False
+        self.api.recover_thumbnail_move()
+        self.assertFalse(self.journal.exists())
+        self.assertFalse(self.folder.exists())
+        self.assertTrue((self.old_store / "a_thumb.webp").is_file())
+
+
+class TestRestart(MoveCase):
+    async def test_a_journal_left_after_the_switch_is_forgotten(self):
+        await self.move({"mode": "folder", "path": str(self.folder)})
+        self.journal.parent.mkdir(parents=True, exist_ok=True)
+        self.journal.write_text(
+            json.dumps(
+                {"state": "copying", "to": {"mode": "folder", "path": str(self.folder)}}
+            ),
+            encoding="utf-8",
+        )
+        self.api._move_recovered = False
+        self.api.recover_thumbnail_move()
+        self.assertFalse(self.journal.exists())
+        self.assertTrue(self.key_files())
+
+    async def test_old_copies_that_waited_are_removed_by_the_finishing_job(self):
+        with patch.object(files, "remove_file", lambda path, **_k: False):
+            _plan, job = await self.move({"mode": "folder", "path": str(self.folder)})
+        self.assertTrue(preview_store.setting().cleanup)
+        self.assertTrue((self.old_store / "a_thumb.webp").exists())
+        # Each removal that failed is an error the job counts and lists by path.
+        rows = job.outcomes["errors"]
+        self.assertEqual(job.counters["errors"], len(rows))
+        self.assertIn("sub/b_thumb.webp", {row["path"] for row in rows})
+        self.assertTrue(all("next start" in row["detail"] for row in rows))
+        job = await self.api.finish_thumbnail_move()
+        await job.task
+        self.assertEqual(job.title, "Move thumbnails (finishing)")
+        self.assertFalse((self.old_store / "a_thumb.webp").exists())
+        self.assertEqual(self.config()["thumbnails"]["location"]["cleanup"], [])
+
+
+class TestAdmission(MoveCase):
+    async def test_automatic_previews_and_setting_changes_wait_for_the_move(self):
+        status, plan = await self.plan({"mode": "folder", "path": str(self.folder)})
+        seen = {}
+        real = self.api._move_copy_one
+
+        def observe(pair, **kwargs):
+            seen["blocker"] = self.api._auto_thumbnail_blocker()
+            return real(pair, **kwargs)
+
+        with patch.object(self.api, "_move_copy_one", observe):
+            resp = await self.client.post(
+                START, json={"mode": "move", "token": plan["token"], "confirmed": True}
+            )
+            job = self.api._jobs.get((await resp.json())["job"]["id"])
+            refused = self.api._thumbnail_change_refused({"THUMBNAIL_QUALITY": 5})
+            self.assertEqual(refused.status, 409)
+            await job.task
+        self.assertIs(seen["blocker"], job)
+
+    def test_the_move_never_trims_memory_and_copies_on_the_job_pool(self):
+        text = "".join(
+            (REPO / "py" / "api" / name).read_text(encoding="utf-8")
+            for name in ("thumbnail_move.py", "thumbnail_prune.py")
+        )
+        self.assertNotIn("_release_job_memory", text)
+        self.assertIn("map_parallel", (REPO / "py/api/thumbnail_move.py").read_text())
+
+
+class TestPrune(MoveCase):
+    async def test_only_this_installs_removed_sources_go(self):
+        await self.move({"mode": "folder", "path": str(self.folder)})
+        install = preview_store.setting().install
+
+        def add(marker):
+            marker["keys"]["old-0123abcd"] = {"identity": "/gone", "install": install}
+            marker["keys"]["theirs-0123abcd"] = {"identity": "/x", "install": "other"}
+
+        preview_location.write_marker(str(self.folder), add)
+        make_png(self.folder / "old-0123abcd" / "x_thumb.webp", size=(4, 4))
+        make_png(self.folder / "theirs-0123abcd" / "y_thumb.webp", size=(4, 4))
+        preview_store.refresh()
+        resp = await self.client.get(PLAN + "?mode=prune")
+        body = await resp.json()
+        self.assertEqual([s["key"] for s in body["sources"]], ["old-0123abcd"])
+        self.assertIn("1 source no longer listed", body["line"])
+        resp = await self.client.post(START, json={"mode": "prune"})
+        job = self.api._jobs.get((await resp.json())["job"]["id"])
+        await job.task
+        self.assertFalse((self.folder / "old-0123abcd").exists())
+        self.assertTrue((self.folder / "theirs-0123abcd" / "y_thumb.webp").exists())
+        keys = preview_location.read_marker(str(self.folder))["keys"]
+        self.assertNotIn("old-0123abcd", keys)
+
+
+class TestRecreate(MoveCase):
+    async def test_recreate_and_its_refusals(self):
+        await self.move({"mode": "folder", "path": str(self.folder)})
+        url = "/prompt_manager/images/thumbnails/location/recreate"
+        resp = await self.client.post(url)
+        self.assertEqual((await resp.json())["code"], "location-has-marker")
+        for child in sorted(self.folder.rglob("*"), reverse=True):
+            child.rmdir() if child.is_dir() else child.unlink()
+        self.folder.rmdir()
+        self.assertFalse(preview_store.refresh())
+        resp = await self.client.post(url)
+        self.assertEqual(resp.status, 200, await resp.text())
+        self.assertTrue(preview_store.current().available)
+        self.assertNotEqual(self.config()["thumbnails"]["location"]["id"], None)
+
+    async def test_use_this_folder_adopts_another_marker(self):
+        await self.move({"mode": "folder", "path": str(self.folder)})
+        marker = preview_location.read_marker(str(self.folder))
+        marker["id"] = "someone-else"
+        (self.folder / preview_location.MARKER_NAME).write_text(
+            json.dumps(marker), encoding="utf-8"
+        )
+        self.assertFalse(preview_store.refresh())
+        resp = await self.client.post(
+            "/prompt_manager/images/thumbnails/location/adopt"
+        )
+        self.assertEqual(resp.status, 200, await resp.text())
+        self.assertTrue(preview_store.current().available)
+        self.assertEqual(self.config()["thumbnails"]["location"]["id"], "someone-else")
+
+
+class TestFiles(unittest.TestCase):
+    def test_mtime_rounds_up_to_two_seconds(self):
+        self.assertEqual(files.rounded_mtime_ns(0), 0)
+        self.assertEqual(files.rounded_mtime_ns(1), files.MTIME_STEP_NS)
+        self.assertEqual(
+            files.rounded_mtime_ns(files.MTIME_STEP_NS), files.MTIME_STEP_NS
+        )
+
+    def test_long_windows_destinations_are_per_file_errors(self):
+        self.assertTrue(files.too_long("D:\\" + "x" * 240, long_paths=False))
+        self.assertFalse(files.too_long("D:\\x", long_paths=False))
+
+    def test_a_failed_copys_temp_that_cannot_go_is_logged(self):
+        with patch.object(files.os, "unlink", side_effect=PermissionError("held")):
+            with self.assertLogs("prompt_manager.thumbnails", "WARNING") as logs:
+                self.assertFalse(files._unlink("D:/thumbs/.pm-move-1.tmp"))
+        self.assertIn(".pm-move-1.tmp", logs.output[0])
+        with patch.object(files.os, "unlink", side_effect=FileNotFoundError()):
+            self.assertTrue(files._unlink("gone"))
+
+    def test_a_held_open_file_is_retried_then_left(self):
+        waits = []
+        error = OSError("in use")
+        error.winerror = 32
+        with patch.object(files.os, "unlink", side_effect=error):
+            self.assertFalse(files.remove_file("x", sleep=waits.append))
+        self.assertEqual(waits, [0.5, 1.0, 2.0])
+
+
+if __name__ == "__main__":
+    unittest.main()
+
+
+class TestReviewFindings(MoveCase):
+    async def test_moving_back_never_removes_previews_a_waiting_cleanup_named(self):
+        await self.move({"mode": "folder", "path": str(self.folder)})
+        other = Path(self.tmpdir) / "thumbs2"
+        # The old copies in the first folder cannot go yet (held open).
+        with patch.object(files, "remove_file", lambda path, **_k: False):
+            await self.move({"mode": "folder", "path": str(other)})
+        self.assertTrue(preview_store.setting().cleanup)
+        _plan, job = await self.move({"mode": "folder", "path": str(self.folder)})
+        self.assertEqual(job.state, "finished", job.log)
+        self.assertEqual(len(self.key_files()), 2)  # the previews moved back stay
+
+    @unittest.skipUnless(hasattr(os, "symlink"), "needs symlinks")
+    def test_empty_folder_removal_never_walks_through_a_link(self):
+        outside = Path(self.tmpdir) / "outside"
+        (outside / "empty").mkdir(parents=True)
+        (outside / "Thumbs.db").write_bytes(b"x")
+        store = Path(self.tmpdir) / "store"
+        store.mkdir()
+        try:
+            (store / "link").symlink_to(outside, target_is_directory=True)
+        except OSError:
+            self.skipTest("symlinks not permitted")
+        files.remove_empty_folders(store)
+        self.assertTrue((outside / "empty").is_dir())
+        self.assertTrue((outside / "Thumbs.db").exists())
+
+    @unittest.skipUnless(hasattr(os, "symlink"), "needs symlinks")
+    async def test_moving_back_never_copies_through_an_existing_linked_folder(self):
+        # Codex P1 on #592: an existing linked folder in the destination was
+        # followed, and a file outside the source was replaced.
+        await self.move({"mode": "folder", "path": str(self.folder)})
+        outside = Path(self.tmpdir) / "outside-p1"
+        outside.mkdir()
+        (outside / "b_thumb.webp").write_bytes(b"not ours")
+        if (self.old_store / "sub").exists():
+            (self.old_store / "sub").rmdir()
+        try:
+            (self.old_store / "sub").symlink_to(outside, target_is_directory=True)
+        except OSError:
+            self.skipTest("symlinks not permitted")
+        _plan, job = await self.move({"mode": "source", "path": None})
+        self.assertEqual((outside / "b_thumb.webp").read_bytes(), b"not ours")
+        self.assertEqual(job.counters["errors"], 1, job.log)
+        self.assertTrue((self.old_store / "a_thumb.webp").is_file())
+
+    @unittest.skipUnless(os.name == "nt", "junctions are Windows only")
+    def test_empty_folder_removal_never_walks_through_a_junction(self):
+        import _winapi
+
+        outside = Path(self.tmpdir) / "outside-j"
+        (outside / "empty").mkdir(parents=True)
+        (outside / "Thumbs.db").write_bytes(b"x")
+        store = Path(self.tmpdir) / "store-j"
+        store.mkdir()
+        _winapi.CreateJunction(str(outside), str(store / "junction"))
+        files.remove_empty_folders(store)
+        self.assertTrue((outside / "empty").is_dir())
+        self.assertTrue((outside / "Thumbs.db").exists())
+
+    async def test_an_unexpected_error_while_copying_changes_nothing(self):
+        def broken(*_args, **_kwargs):
+            raise PermissionError("denied")
+
+        with patch.object(files, "_make_folders", broken):
+            _plan, job = await self.move({"mode": "folder", "path": str(self.folder)})
+        self.assertEqual(job.state, "failed")
+        self.assertFalse(self.journal.exists())
+        self.assertFalse(self.folder.exists())
+        self.assertEqual(preview_store.current().mode, "source")
+
+    async def test_a_journal_that_cannot_be_removed_never_undoes_the_switch(self):
+        from py.api import thumbnail_move
+
+        real = thumbnail_move._remove
+
+        def stuck(path):
+            if str(path) == str(self.journal):
+                raise PermissionError("in use")
+            return real(path)
+
+        with patch.object(thumbnail_move, "_remove", stuck):
+            _plan, job = await self.move({"mode": "folder", "path": str(self.folder)})
+        self.assertEqual(job.state, "finished", job.log)
+        self.assertEqual(len(self.key_files()), 2)
+        self.assertEqual(preview_store.current().mode, "folder")
+
+    async def test_a_hostile_key_is_never_pruned(self):
+        await self.move({"mode": "folder", "path": str(self.folder)})
+        victim = Path(self.tmpdir) / "victim"
+        make_png(victim / "z_thumb.webp", size=(4, 4))
+        marker = preview_location.read_marker(str(self.folder))
+        marker["keys"]["../victim"] = {"identity": "/gone", "install": marker_install()}
+        (self.folder / preview_location.MARKER_NAME).write_text(
+            json.dumps(marker), encoding="utf-8"
+        )
+        preview_store.refresh()
+        resp = await self.client.post(START, json={"mode": "prune"})
+        job = self.api._jobs.get((await resp.json())["job"]["id"])
+        await job.task
+        self.assertTrue((victim / "z_thumb.webp").exists())
+
+    def test_a_hand_edited_source_store_must_be_a_thumbnails_folder(self):
+        elsewhere = Path(self.tmpdir) / "photos"
+        make_png(elsewhere / "keep_thumb.webp", size=(4, 4))
+        self.assertTrue(self.api._remove_store(elsewhere, None, None))
+        self.assertTrue((elsewhere / "keep_thumb.webp").exists())
+
+
+def marker_install():
+    return preview_store.setting().install

tests/trash_route_audit.py

@@ -15,6 +15,8 @@ KNOWN_FILE_READERS = frozenset(
         "images.py:os.open(",
         "__init__.py:Image.open(",
         "logging_routes.py:os.open(",
+        # The thumbnail folder's marker lock file (#320), never a media file.
+        "preview_location.py:os.open(",
         "thumbnail_auto.py:FileResponse(",
         "thumbnail_encode.py:Image.open(",
     }
@@ -77,6 +79,7 @@ ROUTE_AUDIT = {
         "POST /prompt_manager/jobs/autotag",
         "POST /prompt_manager/jobs/thumbnails",
         "GET /prompt_manager/images/thumbnails/plan",
+        "POST /prompt_manager/images/thumbnails/plan",
         "GET /prompt_manager/images/thumbnails/status",
         "POST /prompt_manager/images/generate-thumbnails",
         "POST /prompt_manager/images/generate-thumbnails/progress",
@@ -171,6 +174,8 @@ ROUTE_AUDIT = {
         "POST /prompt_manager/cleanup",
         "POST /prompt_manager/diagnostics/test-link",
         "POST /prompt_manager/images/clear-thumbnails",
+        "POST /prompt_manager/images/thumbnails/location/adopt",
+        "POST /prompt_manager/images/thumbnails/location/recreate",
         "POST /prompt_manager/jobs/model-download",
         "POST /prompt_manager/jobs/{job_id}/cancel",
         "POST /prompt_manager/library",

tools/vendor/frontend-coverage-baseline.json

@@ -641,6 +641,12 @@
     "functions": 95,
     "statements": 98
   },
+  "workspace/system-thumbnail-location.mjs": {
+    "lines": 80,
+    "branches": 65,
+    "functions": 75,
+    "statements": 80
+  },
   "workspace/system-thumbnails.mjs": {
     "lines": 98,
     "branches": 95,

web/js/api-client.js

@@ -62,6 +62,13 @@
                 (typeof job.title === "string" && job.title.length > 0 && job.title.length <= 200))) {
                 this.job = { id: job.id, title: job.title };
             }
+            // A move refused for space (#320) carries its review, so the page
+            // can show what would have moved and why it cannot start.
+            this.plan = null;
+            if (this.code === "insufficient-space" && current && current.plan &&
+                typeof current.plan === "object" && !Array.isArray(current.plan)) {
+                this.plan = current.plan;
+            }
             this.source = null;
             if (this.code === "invalid-source" && current && typeof current.source === "string" &&
                 current.source.length > 0 && current.source.length <= 4096) {

web/js/workspace/activity-job-view.mjs

@@ -170,7 +170,10 @@ export function mountActivityJobView(main, { store, scope, api, jobId }) {
     bar.max = 100;
     const stats = element('dl', undefined, 'stat-row');
     const tileValues = new Map();
-    progressPart.node.append(bar, stats);
+    // #320 D198: a kind with phases (Move thumbnails) says which one runs and what it means, under the bar.
+    const phase = element('p', '', 'small');
+    phase.id = 'job-phase';
+    progressPart.node.append(bar, phase, stats);
 
     const resultsPart = section('Results', 'job-results-title');
     const search = element('label', undefined, 'search-field');
@@ -228,6 +231,7 @@ export function mountActivityJobView(main, { store, scope, api, jobId }) {
     function layout(job) {
         view = viewOf(job);
         if (!view.tiles.length) stats.remove();
+        if (typeof view.phase !== 'function') phase.remove();
         for (const [group, label, tone] of view.tiles) {
             const tile = element('div', undefined, tone ? `stat ${tone}` : 'stat');
             const value = element('dd', '0');
@@ -341,6 +345,7 @@ export function mountActivityJobView(main, { store, scope, api, jobId }) {
             cancel.remove();
         }
         status.textContent = progressLine(job);
+        if (typeof view?.phase === 'function') phase.textContent = view.phase(job);
         // D111: no value while the total is unknown; a job that ended with nothing counted shows how it ended.
         showProgress(bar, job.total > 0 || job.state !== 'finished' ? job : { ...job, progress: 100 });
         bar.setAttribute('aria-label', scan ? 'Scan progress' : `${job.title} progress`);

web/js/workspace/activity-kinds.mjs

@@ -51,6 +51,26 @@ export const removedOf = job => counted(job, 'removed', 'removed');
 /** "12 imported" (System S10: the LoRA reference import). */
 export const importedOf = job => counted(job, 'imported', 'imported');
 
+/** #320 D198: the step a Move thumbnails job is in, and what it means meanwhile (under the progress bar). */
+export function movePhase(job) {
+    const from = job.scope?.from === 'folder' ? 'the old folder' : 'inside each source';
+    if (job.state === 'running') {
+        if (job.scope?.phase === 'remove') return 'Removing the old copies a move left. Thumbnails are read from the new location.';
+        if (job.cancellable !== false) {
+            return `Step 1 of 3: copying. Thumbnails are still read from ${from} until the switch. New thumbnails are not `
+                + 'written until the move ends; originals are shown instead.';
+        }
+        return 'Step 2 and 3 of 3: switching, then removing the old copies. This can no longer be cancelled.';
+    }
+    if (job.state === 'finished') {
+        const left = job.counters?.left ?? 0;
+        return left ? `Thumbnails are now read from the new location. ${number(left)} ${noun(left, 'file')} that `
+            + `${left === 1 ? 'is' : 'are'} not ${left === 1 ? 'a thumbnail was' : 'thumbnails were'} left in place.`
+            : 'Thumbnails are now read from the new location.';
+    }
+    return 'Nothing changed: thumbnails are read from where they were.';
+}
+
 /** "2 output folders": what a scan or a thumbnail job walks. */
 function foldersOf(job) {
     const folders = job.scope?.output_folders;
@@ -159,6 +179,32 @@ export const KINDS = Object.freeze({
         outcomes: tiles([['generated', 'Generated'], ['errors', 'Errors'], ['skipped', 'Skipped']]),
         openable: groups('generated', 'skipped'),
     }),
+    // #320 D198: Move thumbnails, a `thumbnails` job with `scope.mode` "move": Moved (green), Already there, Left in
+    // place and Errors (red); Cancel only until the switch (the server says so through `cancellable`).
+    'thumbnails:move': Object.freeze({
+        ...GENERIC,
+        pillVerb: 'Moving thumbnails',
+        finishToast: job => (job.status_line && job.state !== 'running' ? job.status_line : titleFinished(job)),
+        line: job => (job.state === 'running' ? '' : job.status_line || ''),
+        kept: job => counted(job, 'moved', 'moved'),
+        tiles: tiles([['moved', 'Moved', 'added'], ['skipped', 'Already there', ''], ['left', 'Left in place', ''],
+            ['errors', 'Errors', 'bad']]),
+        outcomes: tiles([['moved', 'Moved'], ['errors', 'Errors'], ['skipped', 'Already there'], ['left', 'Left in place']]),
+        log: 'Move log',
+        phase: movePhase,
+        cancelLabel: 'Cancel move',
+        cancel: 'Cancel move',
+    }),
+    // #320 D200: Remove thumbnails of removed sources (`scope.mode` "prune"): Removed, Left in place and Errors.
+    'thumbnails:prune': Object.freeze({
+        ...GENERIC,
+        pillVerb: 'Removing thumbnails',
+        finishToast: job => (job.status_line && job.state !== 'running' ? job.status_line : titleFinished(job)),
+        line: job => (job.state === 'running' ? '' : job.status_line || ''),
+        kept: removedOf,
+        tiles: tiles([['removed', 'Removed', 'added'], ['left', 'Left in place', ''], ['errors', 'Errors', 'bad']]),
+        outcomes: tiles([['removed', 'Removed'], ['errors', 'Errors'], ['left', 'Left in place']]),
+    }),
     // System S7: a reviewed maintenance plan's apply, titled after its operation; it runs without a known total.
     maintenance: Object.freeze({
         ...GENERIC,
@@ -243,7 +289,9 @@ export const held = job => job.cancellable === false && Object.keys(countsOf(job
  */
 export function viewOf(job) {
     if (held(job)) return GENERIC;
-    const view = kindOf(job.kind);
+    // #320 D198: a thumbnails job chooses its tiles by its scope's mode (move, prune).
+    const mode = job.kind === 'thumbnails' ? job.scope?.mode : null;
+    const view = (mode === 'move' || mode === 'prune') ? KINDS[`thumbnails:${mode}`] : kindOf(job.kind);
     const counts = countsOf(job);
     const has = ([group]) => Object.hasOwn(counts, group);
     if (view.tiles.every(has) && view.outcomes.every(has)) return view;

web/js/workspace/library-location.mjs

@@ -97,6 +97,9 @@ export function reviewGrid(location) {
             ? 'The database backups (config.json stays where PROMPT_MANAGER_CONFIG_PATH names it)'
             : 'config.json and the database backups'],
         ['Old files', 'Renamed *.moved-to-…, never deleted'],
+        // #320 D201: previews kept in one folder do not move with the library (D8A).
+        ...(typeof location.thumbnails_folder === 'string' && location.thumbnails_folder
+            ? [['Thumbnails', `Stay in ${location.thumbnails_folder}. Move them in System › Sources & scans.`]] : []),
     ];
 }
 

web/js/workspace/system-overview.mjs

@@ -28,6 +28,7 @@ import { showWorkspaceToast } from './toast.mjs';
 import { noun } from './plural.mjs';
 import { operationRow } from './system-controls.mjs';
 import { openThumbnailReview } from './thumbnail-review.mjs';
+import { noticeSlot } from './system-thumbnail-location.mjs';
 import { isLocation, locationRow, onMoveEnded } from './library-location.mjs';
 import { STATUS_LOADING, STATUS_URL, backupStatusText } from './system-backups.mjs';
 
@@ -406,7 +407,7 @@ function overviewNodes(overview, { api, activity, notify, scope, navigate, openS
  * `counts` is the shell's sidebar counts (`{get(id), subscribe(listener), freshen(ids)}`, index.mjs): the Images metric
  * shows its `images` count and follows it, and a visit re-reads it as the workspace Overview does (#555).
  */
-export function mountOverview(container, { api, activity, scope, counts = null, navigate = navigateTo,
+export function mountOverview(container, { api, activity, scope, drafts, counts = null, navigate = navigateTo,
     notify = showWorkspaceToast, latestOnly = window.ApiClient.latestOnly, STALE = window.ApiClient.STALE }) {
     const overviewRead = latestOnly({ lifetime: scope.signal });
     const statisticsRead = latestOnly({ lifetime: scope.signal });
@@ -466,7 +467,8 @@ export function mountOverview(container, { api, activity, scope, counts = null,
             if (overview === STALE || scope.signal.aborted) return;
             const { nodes, thumbnails, backup, metrics } = overviewNodes(overview, { api, activity, notify, scope, navigate,
                 openStatistics, images: imagesCount(),
-                openThumbnails: trigger => openThumbnailReview({ api, store: activity, scope, trigger, navigate }) });
+                openThumbnails: trigger => openThumbnailReview({ api, store: activity, scope, trigger, navigate,
+                    location: drafts.get('host', 'thumbnail_location') }) });
             imagesValue = metrics.children[2].querySelector('strong');
             thumbnailRow = thumbnails;
             backupRow = backup;
@@ -522,5 +524,11 @@ export function mountOverview(container, { api, activity, scope, counts = null,
     load();
     loadThumbnails();
     loadBackup();
-    return first;
+    // #320 D199: the unavailable thumbnail folder's notice, first on the page while it lasts.
+    const unavailable = noticeSlot({ api, scope, notify, location: () => drafts.get('host', 'thumbnail_location'),
+        onChange: () => navigate('#/system/sources'), onDone: () => drafts.reloadHost() });
+    scope.cleanup(drafts.subscribe(() => unavailable.paint()));
+    // The anchor goes in first, so the notice (inserted after it) leads the section once the anchor is mounted.
+    unavailable.paint();
+    return [unavailable.anchor, ...first];
 }

web/js/workspace/system-sources.mjs

@@ -35,6 +35,7 @@ import { createButton } from '../shared/toolkit/core.js';
 import { closeWorkspaceDialog, field, openContentDialog } from './workspace-dialogs.mjs';
 import { openMetadataScan } from './activity-scan.mjs';
 import { openThumbnailReview } from './thumbnail-review.mjs';
+import { locationBlock, noticeSlot } from './system-thumbnail-location.mjs';
 import { settingRow, operationRow } from './system-controls.mjs';
 import { HOST_UNAVAILABLE } from './system-appearance.mjs';
 import { showWorkspaceToast } from './toast.mjs';
@@ -234,6 +235,19 @@ export function mountSources(container, { api, activity, scope, drafts, notify =
         nodes.push(notice);
     }
 
+    // #320 D199: the unavailable thumbnail folder's notice, first on the page while it lasts.
+    const thumbnailLocation = () => drafts.get('host', 'thumbnail_location');
+    const location = locationBlock({ api, store: activity, scope, location: thumbnailLocation, notify });
+    const unavailable = noticeSlot({ api, scope, location: thumbnailLocation, notify, onChange: () => location.focus(),
+        onDone: () => drafts.reloadHost?.() });
+    const paintLocation = () => {
+        location.paint();
+        unavailable.paint();
+    };
+    scope.cleanup(drafts.subscribe(paintLocation));
+    nodes.push(unavailable.anchor);
+    paintLocation();
+
     const surface = el('section', undefined, 'surface');
     const heading = el('div', undefined, 'panel-heading');
     const addButton = createButton('+ Add source');
@@ -266,7 +280,10 @@ export function mountSources(container, { api, activity, scope, drafts, notify =
         operationRow({ scope, title: 'Enrich existing metadata', description: 'Resolve both prompts, deduplicate recipes, preserve uncertain links.',
             label: 'Review scan', onClick: scanAll }),
         operationRow({ scope, title: 'Thumbnail generation', description: 'Generate missing previews; rebuilding all is a separate explicit option.',
-            label: 'Review options', onClick: trigger => openThumbnailReview({ api, store: activity, scope, trigger }) }),
+            label: 'Review options', onClick: trigger => openThumbnailReview({ api, store: activity, scope, trigger,
+                location: thumbnailLocation() }) }),
+        // #320 D196: where previews are kept; never a draft (Move thumbnails is its only writer).
+        location.node,
         // D98: the legacy gallery's Refresh, which rereads every source's file list.
         operationRow({ scope, title: 'Refresh files', description: 'Read every source again to find files added or removed outside ComfyUI.',
             label: 'Refresh', onClick: refresh }));

web/js/workspace/system-thumbnail-location.mjs

@@ -0,0 +1,412 @@
+/**
+ * Thumbnail location (#320, beyond the concept: D196–D199, approved static mock `320-mock`): where previews are kept.
+ *
+ * - D196 `locationBlock`: a `.setting-block` in System › Sources & scans after "Thumbnail generation": the radios
+ *   "Inside each source" / "One folder", a "Folder path" field (enabled with One folder, prefilled with the folder in
+ *   use or the server's `default_path`), the status line and "Review move…" (enabled when the choice differs from the
+ *   location in use). Not a host draft: the Save row never saves it; the Move thumbnails job is the only writer (D4B).
+ *   A refusal of the review shows in place under the field (`aria-invalid`), in the server's words.
+ * - D197 `openMoveReview`: the shared review (Move library's shape) over the server's plan
+ *   (`POST /images/thumbnails/plan`): From, To, Thumbnails, Free space, Too long for Windows, Removed after the move,
+ *   Left in place, Links skipped; a folder that is not the default asks for the confirmation checkbox first. "Move
+ *   thumbnails" starts the job (`POST /jobs/thumbnails {mode: "move", token, confirmed}`) and opens it in Activity.
+ * - D199 `locationNotice`: while the thumbnail folder is unavailable, a `.notice` on Overview and Sources & scans
+ *   says why, with "Recreate folder" (a missing folder, after a confirmation) or "Use this folder" (another folder's
+ *   marker), and "Change location".
+ *
+ * The location comes from GET /settings' `thumbnail_location` (`{mode, path, default_path, available, problem,
+ * cleanup_pending}`), read with the host settings.
+ */
+import { createButton } from '../shared/toolkit/core.js';
+import { closeWorkspaceDialog, openContentDialog, openReviewDialog } from './workspace-dialogs.mjs';
+import { busyToast } from './activity-scan.mjs';
+import { showWorkspaceToast } from './toast.mjs';
+import { navigateTo } from './state.mjs';
+import { noun } from './plural.mjs';
+
+export const PLAN_URL = '/prompt_manager/images/thumbnails/plan';
+export const RECREATE_URL = '/prompt_manager/images/thumbnails/location/recreate';
+export const ADOPT_URL = '/prompt_manager/images/thumbnails/location/adopt';
+export const LOCATION_HINT = 'Where previews are kept. Originals are never moved. Avoid cloud-synced folders.';
+export const SENTENCE = 'Every thumbnail is copied and checked before Prompt Manager switches. Scans, deletes and '
+    + 'tagging wait until it finishes; closing this page does not stop it. Originals are never touched.';
+export const REVIEW_FAILED = 'The move could not be reviewed. Try again.';
+export const START_FAILED = 'Move thumbnails could not start. Try again.';
+export const NOTICE_TITLE = 'The thumbnail folder is unavailable';
+const BACK = 'Originals are shown and no thumbnails are written until it is back.';
+const CHECK = 'If its drive is unplugged, connect it; Prompt Manager checks again every few seconds.';
+
+const number = value => Number(value || 0).toLocaleString('en-US');
+const isFolder = location => location?.mode === 'folder';
+
+/** "1.2 GB" (binary units, one decimal, as the server writes them). */
+export function sizeText(bytes) {
+    let size = Number(bytes) || 0;
+    for (const unit of ['B', 'KB', 'MB', 'GB']) {
+        if (size < 1024 || unit === 'GB') return unit === 'B' ? `${size} B` : `${size.toFixed(1)} ${unit}`;
+        size /= 1024;
+    }
+    return `${size.toFixed(1)} GB`;
+}
+
+export const SUGGESTED = 'The suggested folder is next to your library; Review move… shows what happens before anything '
+    + 'is copied.';
+
+/** The status line under the row; `choice` (`{mode, path}`) adds the suggestion while the default folder is chosen. */
+export function statusText(location, choice = null) {
+    if (!location) return '';
+    if (!isFolder(location) && choice?.mode === 'folder' && location.default_path
+        && choice.path?.trim() === location.default_path) {
+        return `Now: inside each source. ${SUGGESTED}`;
+    }
+    let text = isFolder(location) ? `Now: one folder, ${location.path}` : 'Now: inside each source';
+    if (isFolder(location) && location.available === false) return `${text} · unavailable. Originals are shown.`;
+    text += '.';
+    if (isFolder(location) && location.path && location.path === location.default_path) {
+        text += ' Using the default folder next to your library.';
+    }
+    if (location.cleanup_pending > 0) {
+        text += ` Old thumbnails in ${number(location.cleanup_pending)} ${noun(location.cleanup_pending, 'place')} wait to `
+            + 'be removed until it is connected.';
+    }
+    return text;
+}
+
+/** Whether the row's choice (`{mode, path}`) differs from the location in use (Review move… enabled). */
+export function choiceChanged(location, choice) {
+    if (!location) return false;
+    if (choice.mode !== 'folder') return isFolder(location) || Boolean(location.problem);
+    const path = (choice.path || '').trim();
+    return Boolean(path) && (!isFolder(location) || path !== location.path);
+}
+
+/** The D197 review rows (`[term, value]`, a value may be `[lead, items]`) for the server's plan. */
+export function reviewRows(plan) {
+    const totals = plan.totals ?? {};
+    const sources = totals.sources ?? plan.sources?.length ?? 0;
+    const inside = `Inside each source (${number(sources)} ${noun(sources, 'source')})`;
+    const from = plan.from?.mode === 'folder' ? plan.from.path : inside;
+    let to = inside;
+    if (plan.to?.mode === 'folder') {
+        const state = plan.to.default ? (plan.to.will_create ? 'new, next to your library' : 'next to your library')
+            : (plan.to.will_create ? 'new folder' : 'empty folder');
+        to = `${plan.to.path} (${state})`;
+    }
+    const rows = [['From', from], ['To', to],
+        ['Thumbnails', `${number(totals.files)} in ${number(sources)} ${noun(sources, 'source')} · ${sizeText(totals.bytes)}`]];
+    if (Number.isFinite(plan.free_bytes)) rows.push(['Free space', `${sizeText(plan.free_bytes)} free`]);
+    if (totals.too_long > 0) {
+        const more = totals.too_long - (plan.too_long_listed?.length ?? 0);
+        rows.push(['Too long for Windows', [`${number(totals.too_long)} ${noun(totals.too_long, 'thumbnail')} cannot be `
+            + 'stored there. They show the original instead; choose a shorter folder to keep them.',
+        [...(plan.too_long_listed ?? []), ...(more > 0 ? [`and ${number(more)} more`] : [])]]]);
+    }
+    const removed = [];
+    if (totals.orphans > 0) removed.push(`${number(totals.orphans)} ${noun(totals.orphans, 'thumbnail')} whose original is gone`);
+    if (totals.temps > 0) removed.push(`${number(totals.temps)} unfinished temporary ${noun(totals.temps, 'file')}`);
+    for (const source of plan.unlisted ?? []) {
+        removed.push(`1 source no longer listed: ${number(source.files)} ${noun(source.files, 'thumbnail')} are removed; `
+            + 'they are made again if it returns');
+    }
+    const oldCopies = plan.from?.mode === 'folder'
+        ? 'The thumbnail folders of these sources inside it. The folder itself is removed once empty if Prompt Manager created it'
+        : 'The old copies in each source’s thumbnails folder';
+    rows.push(['Removed after the move', [removed.length ? `${oldCopies}, and:` : `${oldCopies}.`, removed]]);
+    if (totals.left_in_place > 0) {
+        rows.push(['Left in place', [`${number(totals.left_in_place)} ${noun(totals.left_in_place, 'file')} that `
+            + `${totals.left_in_place === 1 ? 'is' : 'are'} not ${totals.left_in_place === 1 ? 'a thumbnail' : 'thumbnails'}:`,
+        plan.left_listed ?? []]]);
+    }
+    if (totals.links > 0) {
+        rows.push(['Links skipped', `${number(totals.links)} ${noun(totals.links, 'link')} inside a thumbnails folder `
+            + `${totals.links === 1 ? 'is' : 'are'} not followed`]);
+    }
+    return rows;
+}
+
+/** The notice text for an unavailable `location` (D199). */
+export function noticeText(location) {
+    const where = location?.path || 'The thumbnail folder';
+    switch (location?.problem) {
+    case 'missing': return `${where} is missing. ${BACK} ${CHECK}`;
+    case 'unreachable': return `${where} can’t be reached. ${BACK} ${CHECK}`;
+    case 'marker-mismatch': return `${where} holds another thumbnail folder. ${BACK} Use it if it is yours, or change `
+        + 'the location.';
+    case 'invalid': return `The thumbnail location in config.json can’t be used. ${BACK} Change the location.`;
+    default: return `${where} is being checked. ${BACK}`;
+    }
+}
+
+function el(tag, text, className) {
+    const node = document.createElement(tag);
+    if (text !== undefined) node.textContent = text;
+    if (className) node.className = className;
+    return node;
+}
+
+function radio(value, label, hint) {
+    const node = el('label');
+    const input = el('input');
+    input.type = 'radio';
+    input.name = 'thumb-location';
+    input.value = value;
+    input.id = `loc-${value}`;
+    node.append(input, ` ${label} `, el('span', hint, 'small'));
+    return { node, input };
+}
+
+/** Start the move with the plan's `token`; opens it in Activity, or says why not. */
+export async function startMove({ store, scope, token, confirmed, notify = showWorkspaceToast, navigate = navigateTo,
+    onStale = null }) {
+    try {
+        const result = await store.start('thumbnails', { mode: 'move', token, confirmed });
+        if (scope.signal.aborted) return null;
+        if (result.busy !== undefined) {
+            notify(busyToast(result.busy));
+            return null;
+        }
+        navigate(`#/activity/${result.job.id}`);
+        return result.job;
+    } catch (error) {
+        if (scope.signal.aborted) return null;
+        // The sources or the location changed since the review: review again with what is there now.
+        if (error?.code === 'plan-stale' && onStale) {
+            onStale();
+            return null;
+        }
+        notify(typeof error?.code === 'string' && error.message ? error.message : START_FAILED);
+        return null;
+    }
+}
+
+/** Open the D197 review over `plan`; a plan the server refused for space keeps Move thumbnails unavailable. */
+export function openMoveReview({ scope, store, plan, refusal = null, trigger = null, notify, navigate, onStale = null }) {
+    const options = [];
+    let box = null;
+    if (plan.confirm_required && !refusal) {
+        const label = el('label', undefined, 'review-confirm');
+        box = el('input');
+        box.type = 'checkbox';
+        box.id = 'review-confirmed';
+        label.append(box, el('span', `Prompt Manager may create its own folders and a .promptmanager-thumbnails.json file in ${plan.to.path}.`));
+        options.push(label);
+    }
+    if (refusal) options.push(el('p', refusal, 'small review-refusal'));
+    const review = openReviewDialog({ scope, trigger, title: 'Move thumbnails', grid: reviewRows(plan), options,
+        sentence: refusal ? null : SENTENCE, cancelLabel: 'Cancel', primary: { label: 'Move thumbnails' },
+        onStart: () => { startMove({ store, scope, token: plan.token, confirmed: Boolean(box?.checked), notify, navigate,
+            onStale }); } });
+    const locked = () => Boolean(refusal) || (box !== null && !box.checked);
+    review.setPrimary({ label: 'Move thumbnails', disabled: locked() });
+    if (box) scope.listen(box, 'change', () => review.setPrimary({ label: 'Move thumbnails', disabled: locked() }));
+    return review;
+}
+
+/**
+ * The D196 row; `location()` gives the server's `thumbnail_location` (undefined while read). Returns `{node, paint,
+ * focus}`; `paint()` follows a new location.
+ */
+export function locationBlock({ api, store, scope, location, notify = showWorkspaceToast, navigate = navigateTo }) {
+    const block = el('div', undefined, 'setting-block');
+    block.id = 'location-block';
+    const text = el('div', undefined, 'setting-block-text');
+    const label = el('span', 'Thumbnail location', 'setting-block-label');
+    label.id = 'location-label';
+    text.append(label, el('p', LOCATION_HINT));
+    const choice = el('div', undefined, 'location-choice');
+    choice.setAttribute('role', 'radiogroup');
+    choice.setAttribute('aria-labelledby', 'location-label');
+    const source = radio('source', 'Inside each source', '<source>/thumbnails');
+    const folder = radio('folder', 'One folder', 'one subfolder per source');
+    choice.append(source.node, folder.node);
+    const pathField = el('div', undefined, 'location-path field');
+    const pathLabel = el('label', 'Folder path');
+    pathLabel.htmlFor = 'location-path';
+    const input = el('input');
+    input.id = 'location-path';
+    input.autocomplete = 'off';
+    input.spellcheck = false;
+    const problem = el('p', '', 'small location-problem');
+    problem.id = 'location-problem';
+    problem.setAttribute('role', 'alert');
+    pathField.append(pathLabel, input, problem);
+    const status = el('p', '', 'small location-status');
+    status.id = 'location-status';
+    status.setAttribute('role', 'status');
+    const actions = el('div', undefined, 'row-actions');
+    const review = createButton('Review move…');
+    review.id = 'review-move';
+    actions.append(review);
+    block.append(text, choice, pathField, status, actions);
+
+    let shown;
+    const setProblem = message => {
+        problem.textContent = message || '';
+        pathField.setAttribute('aria-invalid', String(Boolean(message)));
+        // The field itself carries the state, so a screen reader announces it with the reason.
+        if (message) {
+            input.setAttribute('aria-invalid', 'true');
+            input.setAttribute('aria-describedby', 'location-problem');
+        } else {
+            input.removeAttribute('aria-invalid');
+            input.removeAttribute('aria-describedby');
+        }
+    };
+    const current = () => ({ mode: folder.input.checked ? 'folder' : 'source', path: input.value });
+    const repaintButton = () => {
+        const now = location();
+        input.disabled = !folder.input.checked || !now;
+        source.input.disabled = folder.input.disabled = !now;
+        const changed = choiceChanged(now, current());
+        status.textContent = now ? statusText(now, current()) : '';
+        review.setAttribute('aria-disabled', String(!changed));
+        review.classList.toggle('ca-primary', changed);
+    };
+    const paint = () => {
+        const now = location();
+        if (now && now !== shown) {
+            shown = now;
+            (isFolder(now) ? folder : source).input.checked = true;
+            input.value = (isFolder(now) ? now.path : now.default_path) || '';
+            setProblem(null);
+        }
+        repaintButton();
+    };
+    for (const control of [source.input, folder.input]) {
+        scope.listen(control, 'change', () => {
+            if (folder.input.checked && !input.value) input.value = location()?.default_path || '';
+            setProblem(null);
+            repaintButton();
+        });
+    }
+    scope.listen(input, 'input', () => { setProblem(null); repaintButton(); });
+    scope.listen(review, 'click', async () => {
+        if (review.getAttribute('aria-disabled') === 'true') return;
+        const chosen = current();
+        const body = { mode: 'move', location: chosen.mode === 'folder' ? { mode: 'folder', path: chosen.path.trim() }
+            : { mode: 'source' } };
+        review.setAttribute('aria-disabled', 'true');
+        try {
+            const plan = await api.post(PLAN_URL, body, { signal: scope.signal });
+            if (!scope.signal.aborted) {
+                openMoveReview({ scope, store, plan, trigger: review, notify, navigate, onStale: () => review.click() });
+            }
+        } catch (error) {
+            if (scope.signal.aborted) return;
+            if (error?.status === 507 && error?.plan) {
+                openMoveReview({ scope, store, plan: error.plan, refusal: error.plan.insufficient?.text || error.message,
+                    trigger: review });
+            } else if (error?.status === 409 && error?.code === 'busy') {
+                notify(busyToast(error.running ?? null));
+            } else if (typeof error?.code === 'string' && error.code.startsWith('location-') && error.message) {
+                setProblem(error.message);
+                input.focus();
+            } else {
+                notify(typeof error?.message === 'string' && error?.code ? error.message : REVIEW_FAILED);
+            }
+        } finally {
+            repaintButton();
+        }
+    });
+    paint();
+    return {
+        node: block,
+        paint,
+        focus() {
+            folder.input.checked = true;
+            repaintButton();
+            input.focus();
+            block.scrollIntoView?.({ block: 'center' });
+        },
+    };
+}
+
+/** Open the Recreate confirmation for `location`; resolves once it closed. */
+export function openRecreate({ api, scope, location, trigger = null, onDone, notify = showWorkspaceToast }) {
+    const grid = el('dl', undefined, 'meta-grid');
+    for (const [term, value] of [['Folder', location.path], ['Made', 'An empty folder with a new .promptmanager-thumbnails.json'],
+        ['Thumbnails', 'Made again as you browse, or with Thumbnail maintenance']]) {
+        const entry = el('div');
+        entry.append(el('dt', term), el('dd', value));
+        grid.append(entry);
+    }
+    const why = el('p', '', 'small location-problem');
+    const cancel = createButton('Cancel');
+    const go = createButton('Recreate folder', 'ca-primary');
+    const actions = el('div', undefined, 'dialog-actions');
+    actions.append(cancel, go);
+    scope.listen(cancel, 'click', () => closeWorkspaceDialog());
+    scope.listen(go, 'click', async () => {
+        try {
+            const result = await api.post(RECREATE_URL, {}, { signal: scope.signal });
+            if (scope.signal.aborted) return;
+            closeWorkspaceDialog();
+            onDone?.(result.location);
+        } catch (error) {
+            if (scope.signal.aborted) return;
+            why.textContent = error?.message || 'The folder could not be recreated.';
+            if (!error?.message) notify(why.textContent);
+        }
+    });
+    openContentDialog({ scope, title: 'Recreate the thumbnail folder?', trigger, nodes: [grid,
+        el('p', 'If the folder is on a drive that is not connected, connect it instead: recreating would put thumbnails on '
+            + 'another disk.', 'small'), why, actions] });
+}
+
+/**
+ * The D199 notice for `location` (null while it is available): Recreate folder for a missing folder, Use this folder
+ * for another folder's marker, and Change location (`onChange`).
+ */
+export function locationNotice({ api, scope, location, onChange, onDone, notify = showWorkspaceToast }) {
+    if (!isFolder(location) || location.available !== false) return null;
+    const notice = el('div', undefined, 'notice');
+    notice.id = 'unavailable-notice';
+    const text = el('div');
+    text.append(el('h2', NOTICE_TITLE), el('p', noticeText(location)));
+    const actions = el('div', undefined, 'row-actions');
+    if (location.problem === 'missing') {
+        const recreate = createButton('Recreate folder');
+        scope.listen(recreate, 'click', () => openRecreate({ api, scope, location, trigger: recreate, onDone, notify }));
+        actions.append(recreate);
+    } else if (location.problem === 'marker-mismatch') {
+        const adopt = createButton('Use this folder');
+        scope.listen(adopt, 'click', async () => {
+            try {
+                const result = await api.post(ADOPT_URL, {}, { signal: scope.signal });
+                if (!scope.signal.aborted) onDone?.(result.location);
+            } catch (error) {
+                if (!scope.signal.aborted) notify(error?.message || 'The folder could not be used.');
+            }
+        });
+        actions.append(adopt);
+    }
+    const change = createButton('Change location', 'ca-primary');
+    scope.listen(change, 'click', () => onChange?.());
+    actions.append(change);
+    notice.append(text, actions);
+    return notice;
+}
+
+/**
+ * Where a section shows the D199 notice: a comment `anchor` (no element, so the concept comparisons see nothing while the
+ * folder is available) after which `paint()` keeps the notice for `location()`.
+ */
+export function noticeSlot({ api, scope, location, onChange, onDone, notify = showWorkspaceToast }) {
+    const anchor = document.createComment('thumbnail-location-notice');
+    let shownFor;
+    let notice = null;
+    const paint = () => {
+        const now = location();
+        // Not mounted yet: the next paint (a microtask, or the next host read) places it.
+        if (!anchor.parentNode) {
+            queueMicrotask(() => { if (anchor.parentNode && !scope.signal.aborted) paint(); });
+            return;
+        }
+        if (now === shownFor) return;
+        shownFor = now;
+        notice?.remove();
+        notice = locationNotice({ api, scope, location: now, onChange, onDone, notify });
+        if (notice) anchor.after(notice);
+    };
+    return { anchor, paint };
+}

web/js/workspace/system.mjs

@@ -86,7 +86,9 @@ export function systemDrafts(api) {
                 api.get(LORA_STATUS_URL, { cache: 'no-store' }).catch(() => null)]);
             // The store reads a missing value as none and a missing revision as unreadable (no host save). #317: which
             // thumbnail formats the server can write rides along as a read-only value (never drafted or sent).
-            return { values: { ...result.settings, thumbnail_encoders: result.thumbnail_encoders, ...(lora ? loraValues(lora) : {}) },
+            // #320: where previews live rides along the same way (read-only; only Move thumbnails changes it).
+            return { values: { ...result.settings, thumbnail_encoders: result.thumbnail_encoders,
+                thumbnail_location: result.thumbnail_location, ...(lora ? loraValues(lora) : {}) },
                 revision: result.revision };
         },
         async writeHost(changes, { expected_revision }) {

web/js/workspace/thumbnail-review.mjs

@@ -23,6 +23,7 @@ import { busyToast } from './activity-scan.mjs';
 import { showWorkspaceToast } from './toast.mjs';
 import { navigateTo } from './state.mjs';
 import { noun } from './plural.mjs';
+import { sizeText } from './system-thumbnail-location.mjs';
 
 export const THUMBNAIL_PLAN_URL = '/prompt_manager/images/thumbnails/plan';
 export const THUMBNAILS_KIND = 'thumbnails';
@@ -33,7 +34,9 @@ export const PLAN_FAILED = 'The previews could not be counted.';
 export const START_FAILED = 'Thumbnail generation could not start. Try again.';
 export const SETTINGS_UNKNOWN = 'Previews follow the thumbnail settings in System › Appearance.';
 const FORMAT_LABELS = Object.freeze({ webp: 'WebP', jxl: 'JPEG XL', jpeg: 'JPEG' });
-const MODES = new Set(SCOPE_OPTIONS.map(([value]) => value));
+// #320 D200: offered only while previews are kept in one folder (31A).
+export const PRUNE_OPTION = Object.freeze(['prune', 'Remove thumbnails of removed sources']);
+const MODES = new Set([...SCOPE_OPTIONS.map(([value]) => value), PRUNE_OPTION[0]]);
 
 const count = value => value.toLocaleString('en-US');
 const isCount = value => Number.isSafeInteger(value) && value >= 0;
@@ -45,8 +48,22 @@ export function settingsText(plan) {
     return `Previews are written as ${label} · quality ${plan.quality} · ${count(plan.size)} px. Change this in System › Appearance.`;
 }
 
+/** D200: what a prune removes, in `plan.folder` (the thumbnail folder). */
+export function pruneText(plan) {
+    return `Only Prompt Manager’s thumbnails in their folders inside ${plan?.folder || 'the thumbnail folder'} are `
+        + 'removed. If a source comes back, its thumbnails are made again as you browse.';
+}
+
+/** D200: one `[key, "N thumbnails · size"]` per removed source of a prune plan. */
+export function pruneEntries(plan) {
+    const sources = Array.isArray(plan?.sources) ? plan.sources : [];
+    return sources.map(source => [source.key, `${count(source.files)} ${noun(source.files, 'thumbnail')} · `
+        + `${sizeText(source.bytes)}`]);
+}
+
 /** D99: the plan line for the server's `{mode, targets, fresh, remaining}`; an answer without counts failed. */
 export function planText(plan) {
+    if (plan?.mode === 'prune') return typeof plan.line === 'string' && plan.line ? plan.line : PLAN_FAILED;
     if (!plan || !MODES.has(plan.mode) || ![plan.targets, plan.fresh, plan.remaining].every(isCount)) {
         return PLAN_FAILED;
     }
@@ -101,11 +118,18 @@ function select(options, value) {
  * Activity store; `trigger` gets focus back when the dialog closes.
  */
 export function openThumbnailReview({ api, store, scope, trigger = null, mode: initial = 'missing', notify = showWorkspaceToast,
-    navigate = navigateTo }) {
-    const mode = select(SCOPE_OPTIONS, MODES.has(initial) ? initial : 'missing');
+    navigate = navigateTo, location = null }) {
+    const folderMode = location?.mode === 'folder';
+    const options = folderMode ? [...SCOPE_OPTIONS, PRUNE_OPTION] : SCOPE_OPTIONS;
+    const allowed = new Set(options.map(([value]) => value));
+    const mode = select(options, allowed.has(initial) ? initial : 'missing');
     const line = el('p', PLAN_LOADING, 'small');
     line.setAttribute('role', 'status');
     const settings = el('p', SETTINGS_UNKNOWN, 'small');
+    // D200: what the prune removes, under the plan line.
+    const pruneDetail = el('p', '', 'small');
+    const pruneGrid = el('dl', undefined, 'meta-grid');
+    pruneDetail.hidden = pruneGrid.hidden = true;
     const cancel = createButton('Cancel');
     const start = createButton('Start', 'ca-primary');
     const actions = el('div', undefined, 'dialog-actions');
@@ -125,7 +149,20 @@ export function openThumbnailReview({ api, store, scope, trigger = null, mode: i
         // A later choice painted its own line meanwhile, or the dialog closed.
         if (!open || scope.signal.aborted || mode.value !== chosen) return;
         line.textContent = planText(plan);
-        settings.textContent = settingsText(plan);
+        const prune = chosen === 'prune';
+        settings.hidden = prune;
+        const entries = pruneEntries(plan);
+        pruneDetail.hidden = pruneGrid.hidden = !prune || !entries.length;
+        if (prune) {
+            pruneDetail.textContent = pruneText(plan);
+            pruneGrid.replaceChildren(...entries.map(([key, text]) => {
+                const entry = el('div');
+                entry.append(el('dt', key), el('dd', text));
+                return entry;
+            }));
+        } else {
+            settings.textContent = settingsText(plan);
+        }
     }
 
     scope.listen(mode, 'change', paint);
@@ -136,7 +173,7 @@ export function openThumbnailReview({ api, store, scope, trigger = null, mode: i
         startThumbnails({ store, scope, options, notify, navigate });
     });
     openContentDialog({ scope, title: 'Thumbnail maintenance', trigger,
-        nodes: [field('Scope', mode, 'thumb-mode'), line, settings, actions],
+        nodes: [field('Scope', mode, 'thumb-mode'), line, settings, ...(folderMode ? [pruneDetail, pruneGrid] : []), actions],
         onClose: () => { open = false; } });
     paint();
 }

web/js/workspace/workspace-dialogs.mjs

@@ -460,9 +460,13 @@ export function backupNotice({ copy, onBackup }) {
     return notice;
 }
 
+/** A grid value: text, or (#320) `[lead, items]`, a lead sentence and the list under it. */
+const gridValue = value => typeof value === 'string' || (Array.isArray(value) && value.length === 2
+    && typeof value[0] === 'string' && Array.isArray(value[1]) && value[1].every(item => typeof item === 'string'));
+
 function requireGrid(grid) {
     if (!Array.isArray(grid) || grid.length === 0 || grid.some(row => !Array.isArray(row) || row.length !== 2 ||
-        typeof row[0] !== 'string' || !row[0] || typeof row[1] !== 'string')) {
+        typeof row[0] !== 'string' || !row[0] || !gridValue(row[1]))) {
         throw new TypeError('The review needs its [term, value] grid');
     }
 }
@@ -495,7 +499,12 @@ export function openReviewDialog({ scope, title, heading = null, grid, options =
         values.clear();
         list.replaceChildren(...rows.map(([term, value]) => {
             const entry = el('div');
-            const dd = el('dd', value);
+            const dd = el('dd', typeof value === 'string' ? value : value[0]);
+            if (typeof value !== 'string' && value[1].length) {
+                const items = el('ul', undefined, 'review-list');
+                items.append(...value[1].map(item => el('li', item)));
+                dd.append(items);
+            }
             values.set(term, dd);
             entry.append(el('dt', term), dd);
             return entry;

web/js/workspace/workspace.css

@@ -576,3 +576,31 @@ pre.prompt-text { font: inherit; font-size: 12px; line-height: 1.75; margin-bloc
 .review-panel { position: sticky; inset-block-start: 20px; }
 @media (max-width: 700px) { .review-panel { position: static; } }
 @media (min-width: 701px) and (max-height: 720px) { .review-panel { max-block-size: calc(100dvh - 40px); overflow: auto; } }
+
+/* #320 D196: Thumbnail location, a setting wider than a .setting-row column. */
+.setting-block { padding: 17px 0; border-top: 1px solid var(--line); display: grid; gap: 12px; }
+.setting-block-text .setting-block-label { font-size: 12px; font-weight: 500; display: block; }
+.setting-block-text p { font-size: 11px; color: var(--ca-muted); margin-top: 4px; max-width: 640px; }
+.location-choice { display: flex; flex-wrap: wrap; gap: 8px 24px; font-size: 12px; }
+.location-choice label { display: inline-flex; gap: 6px; align-items: baseline; }
+.location-choice .small { color: var(--ca-muted); }
+.location-path { max-width: 640px; margin: 0; }
+.location-path input { width: 100%; min-width: 0; font-family: var(--ca-mono, ui-monospace, monospace); font-size: 12px; }
+.location-path input:disabled { color: var(--ca-muted); }
+.location-path[aria-invalid="true"] input { border-color: var(--red); }
+.location-problem { color: var(--red); margin-top: 6px; }
+.location-problem:empty { display: none; }
+.location-status { margin: 0; overflow-wrap: anywhere; }
+.setting-block .row-actions { justify-content: flex-start; }
+
+/* #320 D197: the move review's lists and confirmation. */
+.review-list { margin: 4px 0 0; padding-inline-start: 18px; font-size: 11px; color: var(--ca-muted); }
+.review-list li { overflow-wrap: anywhere; }
+.review-confirm { display: flex; gap: 8px; align-items: flex-start; font-size: 12px; margin: 12px 0 0;
+    padding: 10px 12px; border: 1px solid var(--line); border-radius: 6px; background: var(--ca-field); }
+.review-confirm input { margin-top: 2px; flex-shrink: 0; }
+.review-refusal { color: var(--red); }
+
+@media (max-width: 700px) {
+    .location-choice { flex-direction: column; }
+}