fix(thumbnails): generate thumbnails automatically and use them everywhere

New outputs never got a preview, so the workspace grid loaded full-size
originals (844 of 19,907 images on a live install, exactly the newest).

- Listings always give images and videos a thumbnail URL; a missing
  preview is made by the serve route on first view (bounded, single
  flight per file) and falls back to the original, remembering files
  that cannot be decoded.
- The image monitor queues each settled output's preview on one
  background worker that follows Worker threads.
- A scan that leaves 200 or more previews missing starts the visible
  thumbnail job; fewer go through the worker.
- Automatic previews wait for a restore, a duplicate removal, a sources
  change, the gallery's clear and exclusive jobs.
- The inspector cover uses the preview; the Overview row counts
  previews that failed.

__init__.py

@@ -132,11 +132,18 @@ try:
     _review_api = globals().get("api")
     if _review_api is not None:
         _global_image_monitor.set_review_recorder(_review_api.record_review_file)
+        # Every settled output's preview is queued for the API's background
+        # worker (#249): the workspace never asks to generate thumbnails.
+        _global_image_monitor.set_thumbnail_queue(_review_api.queue_capture_thumbnail)
         _review_host = getattr(config.server_instance, "app", None)
         if _review_host is not None and hasattr(_review_host, "on_cleanup"):
 
-            async def _stop_review_recording(app, monitor=_global_image_monitor):
+            async def _stop_review_recording(
+                app, monitor=_global_image_monitor, api=_review_api
+            ):
                 monitor.set_review_recorder(None)
+                monitor.set_thumbnail_queue(None)
+                api.close_auto_thumbnails()
 
             _review_host.on_cleanup.insert(0, _stop_review_recording)
 

py/api/__init__.py

@@ -9,6 +9,7 @@ Split into domain-specific mixins for maintainability:
 - JobRoutesMixin: the session job registry behind Activity (py/api/jobs.py)
 - ImageTagJobsMixin: bulk Add tags / Auto-tag as image-tags jobs
 - ThumbnailJobsMixin: thumbnail plan, thumbnail job and the gallery stream on it
+- AutoThumbnailMixin: previews made on first view, on capture and after a scan
 - ReviewRoutesMixin: the Needs review queue, item detail and decisions
 - ReviewRecordingMixin: Needs review recording for the scan and the monitor
 - ReviewEventsMixin: the Needs review change stream an open page re-reads on
@@ -47,6 +48,7 @@ from .settings_routes import SettingsRoutesMixin
 from .job_routes import JobRoutesMixin
 from .image_tag_jobs import ImageTagJobsMixin
 from .thumbnail_jobs import ThumbnailJobsMixin
+from .thumbnail_auto import AutoThumbnailMixin
 from .system import SystemRoutesMixin
 from .maintenance import MaintenanceRoutesMixin
 from .duplicate_files import DuplicateFilesMixin
@@ -289,6 +291,7 @@ class PromptManagerAPI(
     JobRoutesMixin,
     ImageTagJobsMixin,
     ThumbnailJobsMixin,
+    AutoThumbnailMixin,
     SystemRoutesMixin,
     MaintenanceRoutesMixin,
     DuplicateFilesMixin,

py/api/admin.py

@@ -1504,6 +1504,9 @@ class AdminRoutesMixin:
             verb = "cancelled" if state == "cancelled" else "finished"
             job.log_line(f"Scan {verb} · {_plural(job.processed, 'file')} processed")
         job.finish(state)
+        if state == "finished":
+            # After the scan, so the guarded thumbnail job is admitted (#249).
+            await self._hand_off_scan_thumbnails(job, output_dirs)
 
     async def _stream_scan_job(self, request, job):
         """Stream ``job`` as SSE; stop quietly if the client goes away."""

py/api/history.py

@@ -734,7 +734,10 @@ class HistoryRoutesMixin:
                         raise HistoryReadError(422, "response-limit") from None
                 else:
                     prepared = _prepare_current_media(
-                        resolved, scope, thumbnail=kind == "thumbnail"
+                        resolved,
+                        scope,
+                        thumbnail=kind == "thumbnail",
+                        ensure_preview=self._history_preview,
                     )
                 _check_current(scope, self, fence, epoch, generation)
                 return prepared

py/api/history_current.py

@@ -310,8 +310,18 @@ class _OpenedCurrentMedia:
 
 
 def _prepare_current_media(
-    resolved: ResolvedCurrentCandidate, scope: CurrentOutputScope, *, thumbnail: bool
+    resolved: ResolvedCurrentCandidate,
+    scope: CurrentOutputScope,
+    *,
+    thumbnail: bool,
+    ensure_preview=None,
 ) -> _OpenedCurrentMedia:
+    """Open the current file, or its preview for ``thumbnail``.
+
+    A missing preview is made through ``ensure_preview(src, dst, is_video)``
+    (the API's bounded, single-flight automatic preview, #249) when given;
+    when it cannot be made the original is opened, as before.
+    """
     target = resolved.canonical_target
     if thumbnail:
         relative = target.relative_to(resolved.gallery_root)
@@ -328,7 +338,18 @@ def _prepare_current_media(
         try:
             preview.stat()
         except FileNotFoundError:
-            pass
+            # Never written through a preview folder linked out of the root.
+            if (
+                ensure_preview is not None
+                and _canonical(resolved.gallery_root / "thumbnails").is_relative_to(
+                    _canonical(resolved.gallery_root)
+                )
+                and ensure_preview(
+                    target, preview, target.suffix.lower() in VIDEO_EXTENSIONS
+                )
+                == "ready"
+            ):
+                target = _canonical(preview)
         else:
             target = _canonical(preview)
     handle = _open_current_target(target, resolved.namespace, scope)

py/api/images.py

@@ -110,6 +110,8 @@ def _card_header(path: Path) -> tuple:
 
 # The legacy gallery page's thumbnail generation, as Activity lists it (D80).
 THUMBNAIL_GENERATION_TITLE = "Thumbnail generation (gallery page)"
+# Its clear: automatic previews are not written while it runs.
+THUMBNAIL_CLEAR_TITLE = "Thumbnail clear (gallery page)"
 
 
 class _CardFacts(NamedTuple):
@@ -710,20 +712,37 @@ def _iter_media_files(output_path, max_files):
             yield root_path / name
 
 
-def media_thumbnail_url(output_path, rel_path, root_index, *, root_scope=None):
-    """The Images listing's thumbnail URL for *rel_path* under *output_path*,
-    or None when no thumbnail exists (one blocking stat). The Needs review
-    queue cards use the same URL (#249 F, M10)."""
+def media_thumbnail_rel(rel_path):
+    """``thumbnails/<rel>_thumb<ext>`` (POSIX) for the media file *rel_path*.
+
+    The extension is lower case, and a video's preview is a ``.jpg``: the
+    name every thumbnail writer (the plan, the job, automatic previews) uses.
+    """
     rel_path = Path(rel_path)
     extension = rel_path.suffix.lower()
-    thumb_rel = (
+    return (
         f"thumbnails/{rel_path.with_suffix('').as_posix()}_thumb"
         f"{'.jpg' if extension in VIDEO_EXTENSIONS else extension}"
     )
+
+
+def media_thumbnail_url(output_path, rel_path, root_index, *, root_scope=None):
+    """The Images listing's thumbnail URL for *rel_path* under *output_path*
+    (one blocking stat). The Needs review queue cards and the Library use the
+    same URL (#249 F, M10).
+
+    An existing preview's URL carries its mtime as ``v=``. A missing one is
+    still listed (without ``v=``): the serve route makes it on first view and
+    falls back to the original when it cannot (py/api/thumbnail_auto.py), so
+    no card ever loads the full file because its preview was not made yet.
+    """
+    thumb_rel = media_thumbnail_rel(rel_path)
     thumbnail_url = thumbnail_url_for(thumb_rel, Path(output_path) / thumb_rel)
     if thumbnail_url is None:
-        return None
-    thumbnail_url += f"&root={root_index}"
+        quoted = urllib.parse.quote(thumb_rel, safe="/")
+        thumbnail_url = f"/prompt_manager/images/serve/{quoted}?root={root_index}"
+    else:
+        thumbnail_url += f"&root={root_index}"
     if root_scope is not None:
         thumbnail_url += f"&root_scope={root_scope}"
     return thumbnail_url
@@ -1682,6 +1701,11 @@ class ImageRoutesMixin:
                 if image_path.is_file():
                     return self._file_response(image_path)
 
+            # A listed preview that was not made yet is made now (#249).
+            preview = await self._serve_missing_thumbnail(output_dirs, filepath)
+            if preview is not None:
+                return preview
+
             # ComfyUI's cache middleware would otherwise let browsers cache this
             # miss for an hour, hiding a thumbnail generated a moment later.
             return web.json_response(
@@ -1909,6 +1933,8 @@ class ImageRoutesMixin:
                     "total_images": pending + plan["skipped"],
                     "pending": pending,
                     "ready": plan["skipped"],
+                    # Files whose automatic preview failed this session (#249).
+                    "failed": self._auto_thumbnail_failures(),
                     "collection_id": hashlib.sha256(collection.encode()).hexdigest()[
                         :16
                     ],
@@ -2218,7 +2244,7 @@ class ImageRoutesMixin:
         """
         return await self._run_held(
             THUMBNAILS_JOB_KIND,
-            "Thumbnail clear (gallery page)",
+            THUMBNAIL_CLEAR_TITLE,
             "guarded",
             self._clear_thumbnails,
         )

py/api/jobs.py

@@ -515,6 +515,12 @@ class JobRegistry:
     def get(self, job_id):
         return self._jobs.get(job_id)
 
+    def running(self):
+        """The running jobs, oldest first (a snapshot, safe off the event loop)."""
+        jobs = [job for job in tuple(self._jobs.values()) if job.running]
+        jobs.sort(key=lambda job: job._started_seq)
+        return jobs
+
     def active(self, kind):
         """The running job of ``kind``, or None (safe off the event loop)."""
         for job in tuple(self._jobs.values()):

py/api/thumbnail_auto.py

@@ -0,0 +1,576 @@
+"""Automatic thumbnails (#249): previews are a cache the server keeps itself.
+
+The approved mock has no repeated global thumbnail prompt
+(``docs/concepts/workspace/README.md``), so the workspace never asks. Every
+listing gives an image or video a thumbnail URL (``media_thumbnail_url``),
+and the previews behind those URLs are made here, in three ways:
+
+- **On first view.** ``GET /prompt_manager/images/serve/thumbnails/<rel>_thumb
+  <ext>`` for a preview that does not exist yet maps it back to its media file
+  under the same root (text containment first, then a regular file inside the
+  root, never through a link), makes the preview and serves it. At most
+  ``ON_DEMAND_LIMIT`` views write at once and one flight runs per file: a
+  second view of a preview being made waits for it. When the preview cannot be
+  made the original is served (``Cache-Control: no-store``) with its own
+  content type. A file that cannot be decoded is remembered for this session
+  (keyed by its path, size and mtime) so it is not tried again on every view,
+  and counted for the System Overview (``failed`` in
+  ``/images/thumbnails/status``). A preview another program holds open
+  (Windows) is not remembered: the next view tries again.
+- **On capture.** The image monitor hands every new output it settles to
+  ``queue_capture_thumbnail``, which queues its preview for one background
+  worker thread (``AutoThumbnailer``). Each wake takes at most Worker threads
+  (``WORKER_THREADS``) files and writes them on that many threads; enqueuing
+  is O(1), so the monitor and ComfyUI's queue never wait. A file already
+  queued is not queued again, the queue is bounded (``MAX_QUEUED``) and a
+  preview left out is made on its first view instead.
+- **After a scan.** A scan that added files checks the missing previews; from
+  ``SCAN_THUMBNAIL_JOB_THRESHOLD`` up it starts the thumbnail Activity job
+  (missing only, visible and cancellable) instead of writing them silently,
+  and fewer go through the worker.
+
+Every preview is written by ``_generate_one`` with the thumbnail job's default
+size and quality (``DEFAULT_THUMBNAIL_EDGE``, medium), into a temp sibling
+renamed over the old one, so a reader never sees a partial file.
+
+Holds: automatic previews only write files under ``thumbnails/`` and never
+touch the database, like the thumbnail job, so they run beside a scan, image
+tagging and the Needs review recording (S5 M7). They are not written while a
+job runs that replaces the database or removes media and previews: a restore,
+a duplicate removal, a sources change, the gallery's thumbnail clear or an
+``exclusive`` job (``_auto_thumbnail_blocker``). A view then gets the original
+and a queued file is left for its first view.
+"""
+
+import asyncio
+import collections
+import os
+import threading
+from pathlib import Path
+
+from aiohttp import web
+
+from .images import (
+    GALLERY_IMAGE_EXTENSIONS,
+    MEDIA_EXTENSIONS,
+    PREVIEW_IN_USE,
+    PREVIEW_NOT_WRITABLE,
+    THUMBNAIL_CLEAR_TITLE,
+    THUMBNAIL_ENCODINGS,
+    VIDEO_EXTENSIONS,
+    _has_traversal,
+    _is_within,
+    _thumbnail_is_fresh,
+    _worker_threads,
+    media_thumbnail_rel,
+)
+from .jobs import (
+    DUPLICATE_CLEANUP_JOB_KIND,
+    RESTORE_JOB_KIND,
+    THUMBNAILS_JOB_KIND,
+    JobBusy,
+)
+from .settings_routes import SOURCE_CHANGE_JOB_KIND
+from .thumbnail_jobs import DEFAULT_THUMBNAIL_EDGE, ThumbnailOptions
+
+try:
+    from ...utils.parallel import map_parallel
+except ImportError:  # tests import the package top-level
+    from utils.parallel import map_parallel
+
+AUTO_THUMBNAIL_EDGE = DEFAULT_THUMBNAIL_EDGE
+AUTO_THUMBNAIL_SIZE = (AUTO_THUMBNAIL_EDGE, AUTO_THUMBNAIL_EDGE)
+AUTO_THUMBNAIL_ENCODING = THUMBNAIL_ENCODINGS["medium"]
+# Views that may write a preview at the same time; more wait for a slot.
+ON_DEMAND_LIMIT = 3
+# How long a view waits for a slot or another view's flight before it is
+# answered with the original (the next view tries again).
+ON_DEMAND_WAIT_SECONDS = 20
+# Files the capture worker holds; more are made on their first view.
+MAX_QUEUED = 10_000
+# Undecodable files remembered per session (oldest forgotten first).
+MAX_FAILURES_REMEMBERED = 4096
+# Missing previews after a scan from which the visible thumbnail job starts:
+# a fresh install or a newly added source. Fewer go through the worker.
+SCAN_THUMBNAIL_JOB_THRESHOLD = 200
+# Jobs beside which no automatic preview is written (see the module docstring).
+AUTO_REFUSED_KINDS = frozenset(
+    {RESTORE_JOB_KIND, DUPLICATE_CLEANUP_JOB_KIND, SOURCE_CHANGE_JOB_KIND}
+)
+
+READY, FAILED, BUSY, DEFERRED = "ready", "failed", "busy", "deferred"
+# _generate_one errors that say nothing about the file: not remembered.
+_TRANSIENT_ERRORS = frozenset(
+    {PREVIEW_IN_USE, PREVIEW_NOT_WRITABLE, "File no longer exists"}
+)
+
+
+def _fingerprint(src):
+    """``(path, size, mtime_ns)`` of *src*, or None when it cannot be read."""
+    try:
+        stat = os.stat(src)
+    except OSError:
+        return None
+    return (os.path.normcase(str(src)), stat.st_size, stat.st_mtime_ns)
+
+
+def _is_preview_request(filepath):
+    """Text check only: ``thumbnails/<...>/<name>_thumb<lower-case ext>``."""
+    parts = filepath.replace("\\", "/").split("/")
+    if len(parts) < 2 or parts[0] != "thumbnails" or "thumbnails" in parts[1:-1]:
+        return False
+    stem, extension = os.path.splitext(parts[-1])
+    return (
+        stem.endswith("_thumb")
+        and len(stem) > len("_thumb")
+        and extension == extension.lower()
+        and extension in GALLERY_IMAGE_EXTENSIONS
+    )
+
+
+def _regular_file(path):
+    """True for a regular file that is not a link (one lstat)."""
+    try:
+        return not path.is_symlink() and path.is_file()
+    except OSError:
+        return False
+
+
+def _thumbnail_source(root, filepath):
+    """``(src, dst, is_video)`` for the preview *filepath* under *root*, or None.
+
+    *filepath* passed ``_is_preview_request`` and the serve route's traversal
+    check. The source has the preview's name without ``_thumb`` and an
+    extension that previews as the requested one (any case); a ``.jpg`` may
+    also be a video's. It must be a regular file inside *root*. ``dst`` is
+    None when the ``thumbnails`` folder lies outside *root* (a link): the
+    original is then served and nothing is written there.
+    """
+    root = Path(root)
+    parts = filepath.replace("\\", "/").split("/")
+    folder = root.joinpath(*parts[1:-1])
+    stem, extension = os.path.splitext(parts[-1])
+    base = stem[: -len("_thumb")]
+    wanted = [extension] + (sorted(VIDEO_EXTENSIONS) if extension == ".jpg" else [])
+    names = [base + ext for ext in wanted] + [base + ext.upper() for ext in wanted]
+    found = next((folder / n for n in names if _regular_file(folder / n)), None)
+    if found is None:
+        found = _scan_folder(folder, base, set(wanted))
+    if found is None or not _is_within(found, root):
+        return None
+    rel = Path(*parts[1:-1], found.name)
+    dst = root / media_thumbnail_rel(rel)
+    if dst != root.joinpath(*parts):
+        return None
+    if not _is_within(root / "thumbnails", root) or not _is_within(
+        dst, root / "thumbnails"
+    ):
+        dst = None  # a preview folder linked elsewhere is never written
+    return found, dst, found.suffix.lower() in VIDEO_EXTENSIONS
+
+
+def _scan_folder(folder, base, extensions):
+    """A mixed-case extension (``.Png``): one listing of *folder*, or None."""
+    try:
+        with os.scandir(folder) as entries:
+            for entry in entries:
+                stem, extension = os.path.splitext(entry.name)
+                if stem == base and extension.lower() in extensions:
+                    path = Path(entry.path)
+                    if _regular_file(path):
+                        return path
+    except OSError:
+        return None
+    return None
+
+
+class AutoThumbnailer:
+    """Single-flight preview writes for views, and the capture worker.
+
+    ``generate(src, dst, is_video)`` writes one preview and returns
+    ``_generate_one``'s result dict; ``may_write()`` says whether automatic
+    previews may be written now. Thread-safe: views call ``ensure`` on the
+    executor, the monitor calls ``enqueue`` from its own thread.
+    """
+
+    def __init__(
+        self,
+        generate,
+        may_write,
+        logger,
+        *,
+        workers=_worker_threads,
+        on_demand_limit=ON_DEMAND_LIMIT,
+        max_queued=MAX_QUEUED,
+        start_worker=True,
+    ):
+        self._generate = generate
+        self._may_write = may_write
+        self._logger = logger
+        self._workers = workers
+        self._max_queued = max_queued
+        self._start_worker = start_worker
+        self._changed = threading.Condition()
+        self._inflight = {}
+        self._followers = 0
+        self._slot_waiters = 0
+        self._batch = 0
+        self._failures = collections.OrderedDict()
+        self._slots = threading.BoundedSemaphore(on_demand_limit)
+        self._queue = collections.deque()
+        self._queued = set()
+        self._worker = None
+        self._closed = False
+        self._busy = False
+
+    # ── views ─────────────────────────────────────────────────────────
+
+    @property
+    def failure_count(self):
+        with self._changed:
+            return len(self._failures)
+
+    def ensure(self, src, dst, is_video, *, on_demand=True):
+        """Make *dst* for *src* unless it is fresh; returns the outcome.
+
+        ``ready`` (the preview is there), ``failed`` (the file cannot be
+        previewed, remembered), ``busy`` (no slot in time, a transient write
+        error, or another flight that did not finish) or ``deferred`` (a hold
+        refuses automatic writes). Blocking: call it off the event loop.
+        """
+        if _thumbnail_is_fresh(src, dst):
+            return READY
+        fingerprint = _fingerprint(src)
+        if fingerprint is None:
+            return BUSY
+        key = os.path.normcase(str(dst))
+        with self._changed:
+            if fingerprint in self._failures:
+                return FAILED
+            flight = self._inflight.get(key)
+            if flight is None:
+                self._inflight[key] = done = threading.Event()
+            else:
+                self._followers += 1
+                self._changed.notify_all()
+        if flight is not None:
+            try:
+                flight.wait(ON_DEMAND_WAIT_SECONDS)
+            finally:
+                with self._changed:
+                    self._followers -= 1
+            return self._settled(src, dst, fingerprint)
+        try:
+            return self._lead(src, dst, is_video, fingerprint, on_demand)
+        finally:
+            with self._changed:
+                self._inflight.pop(key, None)
+                self._changed.notify_all()
+            done.set()
+
+    def _settled(self, src, dst, fingerprint):
+        if _thumbnail_is_fresh(src, dst):
+            return READY
+        with self._changed:
+            return FAILED if fingerprint in self._failures else BUSY
+
+    def _lead(self, src, dst, is_video, fingerprint, on_demand):
+        if not self._may_write():
+            return DEFERRED
+        if on_demand and not self._take_slot():
+            return BUSY
+        try:
+            if _thumbnail_is_fresh(src, dst):
+                return READY
+            result = self._generate(src, dst, is_video)
+        finally:
+            if on_demand:
+                self._slots.release()
+        if result.get("action") in ("generated", "skipped"):
+            return READY
+        error = result.get("error") or "failed"
+        if error in _TRANSIENT_ERRORS:
+            return BUSY
+        with self._changed:
+            self._failures[fingerprint] = error
+            while len(self._failures) > MAX_FAILURES_REMEMBERED:
+                self._failures.popitem(last=False)
+        self._logger.info(
+            f"No preview for {os.path.basename(str(src))} ({error}); "
+            "the original is shown"
+        )
+        return FAILED
+
+    def _take_slot(self):
+        if self._slots.acquire(blocking=False):
+            return True
+        with self._changed:
+            self._slot_waiters += 1
+            self._changed.notify_all()
+        try:
+            return self._slots.acquire(timeout=ON_DEMAND_WAIT_SECONDS)
+        finally:
+            with self._changed:
+                self._slot_waiters -= 1
+
+    # ── the capture worker ────────────────────────────────────────────
+
+    def enqueue(self, src, dst, is_video):
+        """Queue *dst*'s preview for the worker; False when already queued,
+        full or closed. Never blocks on a write."""
+        key = os.path.normcase(str(dst))
+        with self._changed:
+            if self._closed or key in self._queued:
+                return False
+            if len(self._queue) >= self._max_queued:
+                return False
+            self._queued.add(key)
+            self._queue.append((key, Path(src), Path(dst), bool(is_video)))
+            self._busy = True
+            if self._start_worker and (
+                self._worker is None or not self._worker.is_alive()
+            ):
+                self._worker = threading.Thread(
+                    target=self._run,
+                    name="PromptManagerThumbnailWorker",
+                    daemon=True,
+                )
+                self._worker.start()
+            self._changed.notify_all()
+        return True
+
+    def _run(self):
+        while True:
+            with self._changed:
+                while not self._queue and not self._closed:
+                    self._busy = False
+                    self._changed.notify_all()
+                    self._changed.wait()
+                if self._closed:
+                    self._busy = False
+                    self._changed.notify_all()
+                    return
+                take = min(len(self._queue), max(1, int(self._workers() or 1)))
+                batch = [self._queue.popleft() for _ in range(take)]
+                self._batch = len(batch)
+                self._changed.notify_all()
+            try:
+                map_parallel(self._write_queued, batch, len(batch))
+            finally:
+                with self._changed:
+                    self._batch = 0
+                    for key, *_rest in batch:
+                        self._queued.discard(key)
+
+    def _write_queued(self, item):
+        _key, src, dst, is_video = item
+        try:
+            self.ensure(src, dst, is_video, on_demand=False)
+        except Exception:  # one file never stops the worker
+            self._logger.warning(
+                f"Automatic preview failed for {src.name}", exc_info=True
+            )
+
+    def close(self):
+        """Stop the worker after its current batch; queued files are dropped."""
+        with self._changed:
+            self._closed = True
+            self._queue.clear()
+            self._queued.clear()
+            self._changed.notify_all()
+            worker = self._worker
+        if worker is not None and worker is not threading.current_thread():
+            worker.join(ON_DEMAND_WAIT_SECONDS)
+
+    # ── waits (tests and shutdown; never timers) ──────────────────────
+
+    def _wait_for(self, predicate, timeout):
+        with self._changed:
+            return self._changed.wait_for(predicate, timeout)
+
+    def wait_idle(self, timeout=None):
+        """True once the worker has nothing queued or in hand."""
+        return self._wait_for(lambda: not self._busy, timeout)
+
+    def wait_followers(self, count, timeout=None):
+        """True once *count* views wait on another view's flight."""
+        return self._wait_for(lambda: self._followers >= count, timeout)
+
+    def wait_slot_waiters(self, count, timeout=None):
+        """True once *count* views wait for a write slot."""
+        return self._wait_for(lambda: self._slot_waiters >= count, timeout)
+
+    def wait_batch(self, size, timeout=None):
+        """True once the worker holds a batch of *size* files."""
+        return self._wait_for(lambda: self._batch == size, timeout)
+
+
+class AutoThumbnailMixin:
+    """Automatic previews on the API: views, the monitor's queue, scans."""
+
+    _auto_thumbnailer = None
+    _auto_thumbnailer_lock = threading.Lock()
+
+    def _auto_thumbnails(self):
+        """This API's AutoThumbnailer, created on first use."""
+        with AutoThumbnailMixin._auto_thumbnailer_lock:
+            if self._auto_thumbnailer is None:
+                self._auto_thumbnailer = AutoThumbnailer(
+                    self._write_auto_thumbnail,
+                    lambda: self._auto_thumbnail_blocker() is None,
+                    self.logger,
+                )
+            return self._auto_thumbnailer
+
+    def close_auto_thumbnails(self):
+        """Stop the capture worker (shutdown; safe when it never started)."""
+        thumbnailer = self._auto_thumbnailer
+        if thumbnailer is not None:
+            thumbnailer.close()
+
+    def _auto_thumbnail_failures(self):
+        thumbnailer = self._auto_thumbnailer
+        return 0 if thumbnailer is None else thumbnailer.failure_count
+
+    def _write_auto_thumbnail(self, src, dst, is_video):
+        return self._generate_one(
+            src, dst, AUTO_THUMBNAIL_SIZE, is_video, encoding=AUTO_THUMBNAIL_ENCODING
+        )
+
+    def _auto_thumbnail_blocker(self):
+        """The running job automatic previews wait for, or None (any thread)."""
+        for job in self._jobs.running():
+            if job.admission == "exclusive" or job.kind in AUTO_REFUSED_KINDS:
+                return job
+            if job.kind == THUMBNAILS_JOB_KIND and job.title == THUMBNAIL_CLEAR_TITLE:
+                return job
+        return None
+
+    def _history_preview(self, src, dst, is_video):
+        """A recorded run's missing preview, made like a view's (blocking)."""
+        return self._auto_thumbnails().ensure(Path(src), Path(dst), is_video)
+
+    # ── on first view ─────────────────────────────────────────────────
+
+    def _preview_on_demand(self, root, filepath):
+        """``(outcome, src, dst)`` for a missing preview under *root*, or None."""
+        located = _thumbnail_source(root, filepath)
+        if located is None:
+            return None
+        src, dst, is_video = located
+        if dst is None:
+            return DEFERRED, src, dst
+        outcome = self._auto_thumbnails().ensure(src, dst, is_video)
+        if outcome == READY and not dst.is_file():
+            outcome = BUSY  # removed under us: show the original this once
+        return outcome, src, dst
+
+    async def _serve_missing_thumbnail(self, output_dirs, filepath):
+        """The made preview, or the original when it cannot be made; None
+        when *filepath* names no preview of a media file in *output_dirs*."""
+        if _has_traversal(filepath) or not _is_preview_request(filepath):
+            return None
+        for root in output_dirs:
+            answer = await self._run_in_executor(
+                self._preview_on_demand, Path(root), filepath
+            )
+            if answer is None:
+                continue
+            outcome, src, dst = answer
+            if outcome == READY:
+                return self._file_response(dst)
+            response = web.FileResponse(src)
+            response.headers["Cache-Control"] = "no-store"
+            return response
+        return None
+
+    # ── on capture ────────────────────────────────────────────────────
+
+    def queue_capture_thumbnail(self, image_path):
+        """Queue the preview of a new output (the monitor's thread).
+
+        Returns ``queued``, ``fresh`` (its preview is current) or ``skipped``
+        (not a media file under a current root, a link, a preview itself, or
+        already queued). Never writes on the caller's thread.
+        """
+        located = self._capture_target(image_path)
+        if located is None:
+            return "skipped"
+        src, dst, is_video = located
+        if _thumbnail_is_fresh(src, dst):
+            return "fresh"
+        if self._auto_thumbnails().enqueue(src, dst, is_video):
+            return "queued"
+        return "skipped"
+
+    def _capture_target(self, image_path):
+        path = Path(os.path.abspath(str(image_path)))
+        if path.suffix.lower() not in MEDIA_EXTENSIONS or not _regular_file(path):
+            return None
+        folded = os.path.normcase(str(path))
+        for root in self._get_all_output_dirs():
+            root = Path(os.path.abspath(str(root)))
+            base = os.path.normcase(str(root)).rstrip("\\/") + os.sep
+            if not folded.startswith(base) or not _is_within(path, root):
+                continue
+            rel = Path(os.path.relpath(path, root))
+            if "thumbnails" in rel.parts[:-1]:
+                return None
+            if not _is_within(root / "thumbnails", root):
+                return None
+            dst = root / media_thumbnail_rel(rel)
+            return path, dst, path.suffix.lower() in VIDEO_EXTENSIONS
+        return None
+
+    # ── after a scan ──────────────────────────────────────────────────
+
+    async def _thumbnails_after_scan(self, job, output_dirs=None):
+        """Hand a finished scan's missing previews on; returns what it did.
+
+        None (the scan added nothing), ``job`` (the thumbnail job started for
+        ``SCAN_THUMBNAIL_JOB_THRESHOLD`` or more missing previews), ``busy``
+        (a job refused it; the previews are made on view), ``queued`` (fewer,
+        through the worker) or ``none`` (nothing missing). It counts the
+        sources the scan walked (``output_dirs``, narrowed to the job's
+        ``roots``; every source when None). The server log says which.
+        """
+        if job.counters.get("added", 0) <= 0:
+            return None
+        if output_dirs is None:
+            output_dirs = await self._run_in_executor(self._get_all_output_dirs)
+        roots = job.scope.get("roots")
+        if roots:
+            output_dirs = [output_dirs[i] for i in roots if 0 <= i < len(output_dirs)]
+        plan = await self._run_in_executor(self._plan_gallery_thumbnails, output_dirs)
+        missing = len(plan["targets"]) + plan["remaining"]
+        if missing >= SCAN_THUMBNAIL_JOB_THRESHOLD:
+            options = ThumbnailOptions("missing", AUTO_THUMBNAIL_EDGE, "medium")
+            try:
+                self._start_thumbnail_job(options)
+            except JobBusy as busy:
+                self.logger.info(
+                    f"{missing:,} previews missing after the scan; the thumbnail "
+                    f"job did not start while {busy.running.title} runs"
+                )
+                return "busy"
+            self.logger.info(
+                f"{missing:,} previews missing after the scan; thumbnail job started"
+            )
+            return "job"
+        if not missing:
+            return "none"
+        thumbnailer = self._auto_thumbnails()
+        for src, dst, is_video in plan["targets"]:
+            thumbnailer.enqueue(src, dst, is_video)
+        self.logger.info(f"{missing:,} previews missing after the scan; queued")
+        return "queued"
+
+    async def _hand_off_scan_thumbnails(self, job, output_dirs=None):
+        """``_thumbnails_after_scan`` for the scan runner: never raises."""
+        try:
+            return await self._thumbnails_after_scan(job, output_dirs)
+        except asyncio.CancelledError:
+            raise
+        except Exception:
+            self.logger.warning("Previews after the scan failed", exc_info=True)
+            return None

tests/browser/workspace-images.spec.mjs

@@ -398,18 +398,20 @@ test('preview uses actual Viewer.js and cleans up when closing or leaving Images
 test('real video media has a usable card and native playback cleans up on close and navigation', async ({ page, request }) => {
     await expect(await request.post('/__test__/reset', { data: { include_video: true } })).toBeOK();
     const media = await (await request.get('/prompt_manager/images/output?limit=40')).json();
-    expect(media.images.find(image => image.filename === 'clip.webm').thumbnail_url).toBeNull();
+    // #249: a video with no preview yet still lists its preview URL; the server makes the frame on first view.
+    expect(media.images.find(image => image.filename === 'clip.webm').thumbnail_url).toMatch(/\/thumbnails\/clip_thumb\.jpg\?root=0/);
     await page.goto(workspace);
     await refine(page, { Media: 'video' });
     await expect(cards(page)).toHaveCount(1);
     const card = cards(page).first();
     await expect(card).toHaveAttribute('data-image-key', '0:clip.webm');
-    await expect(card.locator('img')).toHaveCount(0);
     await expect(card.getByRole('button', { name: 'Open clip.webm', exact: true })).toBeVisible();
-    // Legacy video art inside the concept card; D11 names the video's own extension.
-    await expect(card.locator('.media-art > .image-open > video')).toHaveCount(1);
+    // The card shows the preview frame, never the video itself; D11 names the video's own extension.
+    await expect(card.locator('video')).toHaveCount(0);
+    await expect(card.locator('.media-art > .image-open > img')).toHaveCount(1);
     await expect(card.locator('.media-info p')).toHaveText(/(?:^|· )WEBM$/);
-    expect(await card.locator('video').evaluate(node => node.getBoundingClientRect().height)).toBeGreaterThan(20);
+    await expect.poll(() => card.locator('img').evaluate(node => node.complete && node.naturalHeight > 0)).toBe(true);
+    expect(await card.locator('img').evaluate(node => node.getBoundingClientRect().height)).toBeGreaterThan(20);
     await card.getByRole('button', { name: 'Open clip.webm', exact: true }).click();
     const video = page.locator('dialog[open] video');
     await expect(video).toBeVisible();
@@ -713,16 +715,19 @@ test('selected layout stays blue with readable text on hover in either theme', a
     }
 });
 
-test('refresh releases the previous inspector video source', async ({ page, request }) => {
+test('refresh releases the previous inspector cover source', async ({ page, request }) => {
     await expect(await request.post('/__test__/reset', { data: { include_video: true } })).toBeOK();
     await page.goto(workspace);
     const card = cards(page).filter({ hasText: 'clip.webm' });
     await card.getByRole('button', { name: 'Inspect clip.webm', exact: true }).click();
-    await expect(details(page).locator('video')).toBeVisible();
-    await details(page).locator('video').evaluate(element => { window.__oldInspectorVideo = element; });
+    // #249: a video's cover is its preview frame, not the video.
+    const cover = details(page).locator('img.detail-cover');
+    await expect(cover).toHaveAttribute('src', /\/thumbnails\/clip_thumb\.jpg\?/);
+    await expect(details(page).locator('video')).toHaveCount(0);
+    await cover.evaluate(element => { window.__oldInspectorCover = element; });
     await rescan(page);
-    await expect.poll(() => page.evaluate(() => !window.__oldInspectorVideo.isConnected)).toBe(true);
-    expect(await page.evaluate(() => window.__oldInspectorVideo.getAttribute('src'))).toBeNull();
+    await expect.poll(() => page.evaluate(() => !window.__oldInspectorCover.isConnected)).toBe(true);
+    expect(await page.evaluate(() => window.__oldInspectorCover.getAttribute('src'))).toBeNull();
 });
 
 test('Import Rescan rescans cached files and preserves the collection when rescan fails', async ({ page, request }) => {

tests/browser/workspace-thumbnails.spec.mjs

@@ -0,0 +1,106 @@
+import { startCoverage, saveCoverage } from '../../tools/vendor/browser-coverage.mjs';
+import { test, expect } from './fixtures.mjs';
+
+/**
+ * #249: cards and strips load previews, never the full file. The approved mock has no repeated thumbnail prompt
+ * (docs/concepts/workspace/README.md), so a preview that was not made yet is made by the server on first view:
+ * the listing still names the `_thumb` URL. Only the media viewer's main image (and downloads) load the original.
+ */
+const images = '/prompt_manager/workspace#/images';
+const library = '/prompt_manager/workspace#/library';
+const cards = page => page.locator('article[data-image-key]');
+const details = page => page.getByRole('complementary', { name: 'Selected image details', exact: true });
+const SERVE = '/prompt_manager/images/serve/';
+
+/** Every media file the page asks the server for, as `{path, preview}` (path relative to its root). */
+function recordMedia(page) {
+    const seen = [];
+    page.on('request', request => {
+        const url = new URL(request.url());
+        if (!url.pathname.startsWith(SERVE)) return;
+        const path = decodeURIComponent(url.pathname.slice(SERVE.length));
+        seen.push({ path, preview: /^thumbnails\/.+_thumb\.[a-z0-9]+$/.test(path) });
+    });
+    return seen;
+}
+/** Every card image has loaded (or failed): what it asked for is all in `seen`. */
+const settled = page => page.evaluate(() => Promise.all([...document.querySelectorAll('article img')].map(image => image.decode().catch(() => {}))));
+const originals = seen => seen.filter(item => !item.preview).map(item => item.path);
+
+test.beforeEach(async ({ page, browserName }) => {
+    await startCoverage(page, browserName);
+    page.thumbErrors = [];
+    page.on('pageerror', error => page.thumbErrors.push(error.message));
+});
+test.afterEach(async ({ page, browserName }) => {
+    await saveCoverage(page, browserName);
+    expect(page.thumbErrors).toEqual([]);
+});
+
+test('a page of the Images grid requests previews, never originals, in every layout and size', async ({ page, request }) => {
+    await expect(await request.post('/__test__/reset')).toBeOK();
+    // Newest outputs with no preview yet: the live finding (844 such files listed with a null thumbnail).
+    await expect(await request.post('/__test__/system/unthumbnailed', { data: { count: 6 } })).toBeOK();
+    const listing = await (await request.get('/prompt_manager/images/output?limit=40')).json();
+    const fresh = listing.images.filter(image => image.path.startsWith('unthumbnailed/'));
+    expect(fresh).toHaveLength(6);
+    for (const image of fresh) expect(image.thumbnail_url).toMatch(/\/thumbnails\/unthumbnailed\/plain_\d{3}_thumb\.png\?root=0/);
+    const seen = recordMedia(page);
+    await page.goto(images);
+    await expect(cards(page)).toHaveCount(8);
+    await settled(page);
+    expect(originals(seen)).toEqual([]);
+    expect(seen.filter(item => item.path.startsWith('thumbnails/unthumbnailed/'))).toHaveLength(6);
+    // The server made them on that first view.
+    await expect.poll(async () => (await (await request.get('/prompt_manager/images/thumbnails/status')).json()).pending).toBe(0);
+    for (const layout of ['List', 'Masonry', 'Grid']) {
+        await page.getByRole('button', { name: layout, exact: true }).click();
+        await expect(page.getByRole('button', { name: layout, exact: true })).toHaveAttribute('aria-pressed', 'true');
+        for (const density of ['large', 'compact', 'comfortable']) {
+            await page.getByRole('combobox', { name: 'Thumbnail size', exact: true }).selectOption(density);
+            await expect(cards(page)).toHaveCount(8);
+            await settled(page);
+        }
+    }
+    expect(originals(seen)).toEqual([]);
+});
+
+test('the inspector cover is the preview and only the media viewer loads the original', async ({ page, request }) => {
+    await expect(await request.post('/__test__/reset')).toBeOK();
+    await expect(await request.post('/__test__/system/unthumbnailed', { data: { count: 1 } })).toBeOK();
+    const seen = recordMedia(page);
+    await page.goto(images);
+    const card = cards(page).filter({ hasText: 'plain_001.png' });
+    await card.getByRole('button', { name: 'Inspect plain_001.png', exact: true }).click();
+    const cover = details(page).locator('img.detail-cover');
+    await expect(cover).toHaveAttribute('src', /\/thumbnails\/unthumbnailed\/plain_001_thumb\.png\?/);
+    await settled(page);
+    await cover.evaluate(image => image.decode().catch(() => {}));
+    expect(originals(seen)).toEqual([]);
+    await card.getByRole('button', { name: 'Open plain_001.png', exact: true }).click();
+    await expect(page.locator('dialog#viewer-dialog')).toBeVisible();
+    await expect.poll(() => originals(seen)).toEqual(['unthumbnailed/plain_001.png']);
+    await page.getByRole('button', { name: 'Close image viewer', exact: true }).click();
+});
+
+test('a video card shows its preview frame instead of loading the video', async ({ page, request }) => {
+    await expect(await request.post('/__test__/reset', { data: { include_video: true } })).toBeOK();
+    const seen = recordMedia(page);
+    await page.goto(images);
+    const card = cards(page).filter({ hasText: 'clip.webm' });
+    await expect(card.locator('.media-art > .image-open > img')).toHaveAttribute('src', /\/thumbnails\/clip_thumb\.jpg\?/);
+    await expect(card.locator('video')).toHaveCount(0);
+    await settled(page);
+    expect(originals(seen)).toEqual([]);
+});
+
+test('Library cards and the shared gallery request previews only', async ({ page, request }) => {
+    await expect(await request.post('/__test__/reset', { data: { native_library: true } })).toBeOK();
+    const seen = recordMedia(page);
+    await page.goto(library);
+    const cat = page.locator('article[data-prompt-id]').filter({ has: page.getByText('Cat portrait', { exact: true }) });
+    await expect(cat.locator('.card-art img')).not.toHaveCount(0);
+    await settled(page);
+    expect(seen.length).toBeGreaterThan(0);
+    expect(originals(seen)).toEqual([]);
+});

tests/js/workspace-history-media.test.js

@@ -39,3 +39,4 @@ test('stale inspection echoes use the canonical recovery vocabulary and identity
         });
     }
 });
+test('the recorded output inspector cover is its preview, not the original (#249)',()=>{const a=adapter(),target=media('2',1);const record=a.toRecord(envelope(target),target,extractGenerationFacts);assert.deepEqual(record.preview,{url:target.thumbnail_url});assert.notEqual(record.preview.url,record.download.url);});

tests/js/workspace-inspection-state.test.js

@@ -895,3 +895,20 @@ test('cache off never rereads a selection whose source changed, and refresh neve
     assert.equal(deferred.state.deferred, true);
     deferred.dispose();
 });
+
+test('record carries the listing thumbnail as its preview, never another file (#249)', () => {
+    const toRecord = contract('toInspectionRecord'), entry = image();
+    const thumb = `/prompt_manager/images/serve/thumbnails/folder/a%20%23%25_thumb.png?root=0&root_scope=${scopeA}`;
+    assert.deepEqual(toRecord(envelope(entry), { ...entry, thumbnail_url: thumb }, extractGenerationFacts).preview, { url: thumb });
+    const versioned = `/prompt_manager/images/serve/thumbnails/folder/a%20%23%25_thumb.png?v=12&root=0&root_scope=${scopeA}`;
+    assert.deepEqual(toRecord(envelope(entry), { ...entry, thumbnail_url: versioned }, extractGenerationFacts).preview, { url: versioned });
+    // No thumbnail, or one naming another file, root or origin: no preview (the cover keeps the original).
+    for (const thumbnail_url of [undefined, null, '', thumb.replace('a%20%23%25_thumb', 'b_thumb'), thumb.replace('root=0', 'root=1'),
+        thumb.replace(scopeA, scopeB), `https://other.test${thumb}`, thumb.replace('thumbnails/', ''), `${thumb}#x`]) {
+        assert.equal(toRecord(envelope(entry), { ...entry, thumbnail_url }, extractGenerationFacts).preview, null, String(thumbnail_url));
+    }
+    const video = { root_index: 0, path: 'clip.MP4', filename: 'clip.MP4', url: `/prompt_manager/images/serve/clip.MP4?root=0&root_scope=${scopeA}`, root_scope: scopeA,
+        thumbnail_url: `/prompt_manager/images/serve/thumbnails/clip_thumb.jpg?root=0&root_scope=${scopeA}` };
+    const data = envelope(video); data.file = { ...data.file, media_type: 'video', extension: '.mp4', format: 'MP4' };
+    assert.deepEqual(toRecord(data, video, extractGenerationFacts).preview, { url: video.thumbnail_url });
+});

tests/js/workspace-system-overview.test.js

@@ -253,3 +253,11 @@ test('waiting stops when its signal aborts', async () => {
     gone.abort(new Error('already'));
     await assert.rejects(waitForJob(fakeActivity(), JOB, { signal: gone.signal }), { message: 'already' });
 });
+
+test('the thumbnail row counts previews that could not be made (#249)', async () => {
+    const { thumbnailTitle } = await overviewModule;
+    assert.equal(thumbnailTitle({ pending: 12, failed: 2 }), '12 missing thumbnails · 2 failed');
+    assert.equal(thumbnailTitle({ pending: 1, failed: 1 }), '1 missing thumbnail · 1 failed');
+    assert.equal(thumbnailTitle({ pending: 0, failed: 0 }), 'No missing thumbnails');
+    assert.equal(thumbnailTitle({ pending: 3, failed: 'x' }), '3 missing thumbnails');
+});

tests/test_api_history_outputs.py

@@ -401,7 +401,7 @@ class HistoryOutputAPITests(ImageAPITestCase):
         self.assertEqual(response.status, 400)
         await response.read()
 
-    async def test_media_ranges_head_and_existing_thumbnail_without_generation(self):
+    async def test_media_ranges_head_and_a_preview_made_on_first_view(self):
         path = self.output_dir / "clip.mp4"
         payload = bytes(range(256)) * 600
         path.write_bytes(payload)
@@ -430,9 +430,14 @@ class HistoryOutputAPITests(ImageAPITestCase):
         self.assertEqual(response.status, 200)
         self.assertEqual(int(response.headers["Content-Length"]), len(payload))
         self.assertEqual(await response.read(), b"")
+        # A missing preview is made on first view (#249): an undecodable video
+        # gets the placeholder frame, never the whole file.
+        preview = self.output_dir / "thumbnails" / "clip_thumb.jpg"
         response = await self.current(descriptor["thumbnail_url"])
-        self.assertEqual(await response.read(), payload)
-        preview = make_png(self.output_dir / "thumbnails" / "clip_thumb.jpg")
+        self.assertEqual(response.headers["Content-Type"], "image/jpeg")
+        self.assertEqual(await response.read(), preview.read_bytes())
+        self.assertLess(preview.stat().st_size, len(payload))
+        make_png(preview)
         response = await self.current(descriptor["thumbnail_url"])
         self.assertEqual(await response.read(), preview.read_bytes())
 

tests/test_api_review.py

@@ -397,15 +397,20 @@ class TestQueueReads(ReviewRoutesTestCase):
         self._no_paths()
 
     async def test_queue_cards_use_the_images_listing_thumbnail(self):
-        # M10: a known thumbnail is the card's media, as on the Images page;
-        # without one the card falls back to the original.
+        # M10: the card's media is the Images listing's thumbnail. Without a
+        # preview yet it is still the thumbnail URL (made on first view, #249).
         self._require_routes()
         from py.api.images import _output_image_entry
 
         path = self._ambiguous("sub/f0.png")
         self._record(path)
         item = (await self._list())["items"][0]
-        self.assertIsNone(item["thumbnail_url"])
+        scope = item["target"]["root_scope"]
+        self.assertEqual(
+            item["thumbnail_url"],
+            "/prompt_manager/images/serve/thumbnails/sub/f0_thumb.png"
+            f"?root=0&root_scope={scope}",
+        )
         thumb = self.output_dir / "thumbnails" / "sub" / "f0_thumb.png"
         thumb.parent.mkdir(parents=True)
         write_png(thumb, prompt=ambiguous_graph("cat", "dog"), color=(1, 2, 3))

tests/test_thumbnail_auto.py

@@ -0,0 +1,634 @@
+"""Automatic thumbnails (#249): previews are a cache the server keeps itself.
+
+The approved mock (docs/concepts/workspace/README.md) has no repeated global
+"generate thumbnails" prompt, so the workspace never asks:
+
+- every listing gives an image or video a thumbnail URL, existing or not;
+- the serve route makes a missing preview on first view (bounded, single
+  flight per file) and falls back to the original when it cannot, remembering
+  the failure so a broken file is not retried on every view;
+- the image monitor queues each new output's preview on one background worker;
+- a scan that leaves many previews missing starts the visible thumbnail job.
+
+Ordering is driven by threading/asyncio events, never by timers.
+"""
+
+import asyncio
+import os
+import sys
+import threading
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+try:
+    from tests.test_api_images import ThumbnailTestCase, make_png
+except ImportError:  # discovered with tests/ as the top-level directory
+    from test_api_images import ThumbnailTestCase, make_png
+
+from PIL import Image  # noqa: E402
+
+import py.api.images as images_module  # noqa: E402
+import py.api.thumbnail_auto as auto  # noqa: E402
+from py.api.jobs import THUMBNAILS_JOB_KIND  # noqa: E402
+
+SERVE = "/prompt_manager/images/serve/"
+WAIT = 10
+
+
+def _entry(api, root, rel):
+    return images_module._output_image_entry(root / rel, root, 0)
+
+
+class AutoThumbnailCase(ThumbnailTestCase):
+    async def asyncSetUp(self):
+        await super().asyncSetUp()
+        self.addCleanup(self.api.close_auto_thumbnails)
+
+    def _count_writes(self):
+        """Count every preview write (and its thread) through _generate_one."""
+        calls = []
+        lock = threading.Lock()
+        original = self.api._generate_one
+
+        def counting(src, *args, **kwargs):
+            with lock:
+                calls.append(Path(src).name)
+            return original(src, *args, **kwargs)
+
+        self.api._generate_one = counting
+        return calls
+
+    def _gate_writes(self):
+        """Hold every preview write until released: (entered, release, active)."""
+        entered = threading.Event()
+        release = threading.Event()
+        state = {"active": 0, "peak": 0}
+        lock = threading.Lock()
+        original = self.api._generate_one
+
+        def gated(*args, **kwargs):
+            with lock:
+                state["active"] += 1
+                state["peak"] = max(state["peak"], state["active"])
+            entered.set()
+            try:
+                release.wait(WAIT)
+                return original(*args, **kwargs)
+            finally:
+                with lock:
+                    state["active"] -= 1
+
+        self.api._generate_one = gated
+        self.addCleanup(release.set)
+        return entered, release, state
+
+    async def _wait(self, event):
+        loop = asyncio.get_running_loop()
+        self.assertTrue(await loop.run_in_executor(None, event.wait, WAIT))
+
+    async def _idle(self):
+        loop = asyncio.get_running_loop()
+        thumbnailer = self.api._auto_thumbnails()
+        self.assertTrue(await loop.run_in_executor(None, thumbnailer.wait_idle, WAIT))
+
+    def _hold(self, kind, title="Database restore (admin page)", admission="records"):
+        """A running, non-cancellable registry hold of ``kind`` until cleanup."""
+        gate = asyncio.Event()
+
+        async def runner(job):
+            await gate.wait()
+
+        job = self.api._jobs.start(
+            kind, title, admission=admission, cancellable=False, runner=runner
+        )
+
+        async def release():
+            gate.set()
+            await asyncio.wait_for(job.task, 5)
+
+        self.addAsyncCleanup(release)
+        return job
+
+
+class TestListingsAlwaysCarryAThumbnailUrl(AutoThumbnailCase):
+    def test_missing_image_preview_still_gets_a_thumbnail_url(self):
+        make_png(self.output_dir / "sub" / "new.png", size=(64, 64))
+        entry = _entry(self.api, self.output_dir, "sub/new.png")
+        self.assertEqual(
+            entry["thumbnail_url"],
+            f"{SERVE}thumbnails/sub/new_thumb.png?root=0",
+        )
+        self.assertFalse((self.output_dir / "thumbnails").exists(), "no write")
+
+    def test_missing_video_preview_is_a_jpeg_url(self):
+        (self.output_dir / "clip.mp4").write_bytes(b"not really a video")
+        entry = _entry(self.api, self.output_dir, "clip.mp4")
+        self.assertEqual(
+            entry["thumbnail_url"], f"{SERVE}thumbnails/clip_thumb.jpg?root=0"
+        )
+
+    def test_existing_preview_keeps_its_mtime_version(self):
+        make_png(self.output_dir / "a.png", size=(64, 64))
+        make_png(self.output_dir / "thumbnails" / "a_thumb.png")
+        entry = _entry(self.api, self.output_dir, "a.png")
+        self.assertRegex(
+            entry["thumbnail_url"],
+            r"^/prompt_manager/images/serve/thumbnails/a_thumb\.png\?v=\d+&root=0$",
+        )
+
+    async def test_workspace_listing_never_returns_a_null_thumbnail(self):
+        for name in ("one.png", "two.png", "three.png"):
+            make_png(self.output_dir / name, size=(64, 64))
+        resp = await self.client.get("/prompt_manager/images/output?limit=10")
+        data = await resp.json()
+        self.assertTrue(data["success"], data)
+        urls = [image["thumbnail_url"] for image in data["images"]]
+        self.assertEqual(len(urls), 3)
+        for url in urls:
+            self.assertIn("_thumb.png", url)
+
+
+class TestThumbnailOnFirstView(AutoThumbnailCase):
+    async def test_missing_preview_is_generated_and_served(self):
+        make_png(self.output_dir / "sub" / "big.png", size=(1200, 900))
+        url = _entry(self.api, self.output_dir, "sub/big.png")["thumbnail_url"]
+
+        resp = await self.client.get(url)
+
+        self.assertEqual(resp.status, 200)
+        self.assertEqual(resp.headers["Content-Type"], "image/png")
+        body = await resp.read()
+        thumb = self.output_dir / "thumbnails" / "sub" / "big_thumb.png"
+        self.assertTrue(thumb.is_file())
+        self.assertEqual(body, thumb.read_bytes())
+        with Image.open(thumb) as img:
+            self.assertEqual(max(img.size), auto.AUTO_THUMBNAIL_EDGE)
+        # No temp sibling is ever left next to the preview.
+        self.assertEqual(os.listdir(thumb.parent), ["big_thumb.png"])
+
+    async def test_upper_case_extension_source_is_found(self):
+        make_png(self.output_dir / "Shot.PNG", size=(64, 64))
+        url = _entry(self.api, self.output_dir, "Shot.PNG")["thumbnail_url"]
+        self.assertIn("thumbnails/Shot_thumb.png", url)
+        resp = await self.client.get(url)
+        self.assertEqual(resp.status, 200)
+        self.assertTrue((self.output_dir / "thumbnails" / "Shot_thumb.png").is_file())
+
+    async def test_video_preview_is_generated_as_jpeg(self):
+        (self.output_dir / "clip.mp4").write_bytes(b"not really a video")
+        url = _entry(self.api, self.output_dir, "clip.mp4")["thumbnail_url"]
+        resp = await self.client.get(url)
+        self.assertEqual(resp.status, 200)
+        self.assertEqual(resp.headers["Content-Type"], "image/jpeg")
+        self.assertTrue((self.output_dir / "thumbnails" / "clip_thumb.jpg").is_file())
+
+    async def test_concurrent_views_generate_once(self):
+        make_png(self.output_dir / "a.png", size=(256, 256))
+        url = _entry(self.api, self.output_dir, "a.png")["thumbnail_url"]
+        calls = self._count_writes()
+        entered, release, _state = self._gate_writes()
+
+        first = asyncio.ensure_future(self.client.get(url))
+        await self._wait(entered)
+        others = [asyncio.ensure_future(self.client.get(url)) for _ in range(4)]
+        # Every follower is waiting on the leader's flight before it is released.
+        thumbnailer = self.api._auto_thumbnails()
+        loop = asyncio.get_running_loop()
+        self.assertTrue(
+            await loop.run_in_executor(None, thumbnailer.wait_followers, 4, WAIT)
+        )
+        release.set()
+        responses = [await first] + [await task for task in others]
+
+        self.assertEqual([r.status for r in responses], [200] * 5)
+        self.assertEqual(calls, ["a.png"])
+
+    async def test_on_demand_generation_is_bounded(self):
+        names = [f"p{i}.png" for i in range(auto.ON_DEMAND_LIMIT + 2)]
+        for name in names:
+            make_png(self.output_dir / name, size=(64, 64))
+        entered, release, state = self._gate_writes()
+        tasks = [
+            asyncio.ensure_future(
+                self.client.get(_entry(self.api, self.output_dir, n)["thumbnail_url"])
+            )
+            for n in names
+        ]
+        await self._wait(entered)
+        thumbnailer = self.api._auto_thumbnails()
+        loop = asyncio.get_running_loop()
+        # The extra requests queue for a slot instead of writing beside them.
+        self.assertTrue(
+            await loop.run_in_executor(None, thumbnailer.wait_slot_waiters, 2, WAIT)
+        )
+        self.assertEqual(state["active"], auto.ON_DEMAND_LIMIT)
+        release.set()
+        responses = [await task for task in tasks]
+        self.assertEqual([r.status for r in responses], [200] * len(names))
+        self.assertLessEqual(state["peak"], auto.ON_DEMAND_LIMIT)
+
+    async def test_broken_file_falls_back_to_the_original_and_is_remembered(self):
+        broken = self.output_dir / "broken.png"
+        broken.write_bytes(b"\x89PNG\r\n\x1a\n not an image")
+        url = _entry(self.api, self.output_dir, "broken.png")["thumbnail_url"]
+        calls = self._count_writes()
+
+        first = await self.client.get(url)
+        second = await self.client.get(url)
+
+        for resp in (first, second):
+            self.assertEqual(resp.status, 200)
+            self.assertEqual(resp.headers["Content-Type"], "image/png")
+            self.assertEqual(resp.headers["Cache-Control"], "no-store")
+            self.assertEqual(await resp.read(), broken.read_bytes())
+        self.assertEqual(calls, ["broken.png"], "a known failure is not retried")
+        self.assertFalse((self.output_dir / "thumbnails" / "broken_thumb.png").exists())
+
+        status = await (
+            await self.client.get("/prompt_manager/images/thumbnails/status")
+        ).json()
+        self.assertEqual(status["failed"], 1)
+
+        # A changed file is tried again.
+        later = broken.stat().st_mtime + 5
+        make_png(broken, size=(64, 64))
+        os.utime(broken, (later, later))
+        third = await self.client.get(url)
+        self.assertEqual(third.status, 200)
+        self.assertEqual(calls, ["broken.png", "broken.png"])
+        self.assertTrue((self.output_dir / "thumbnails" / "broken_thumb.png").is_file())
+
+    async def test_locked_preview_is_not_remembered_as_a_failure(self):
+        make_png(self.output_dir / "a.png", size=(64, 64))
+        url = _entry(self.api, self.output_dir, "a.png")["thumbnail_url"]
+        locked = PermissionError(13, "in use")
+        locked.winerror = 32
+
+        with patch.object(images_module.os, "replace", side_effect=locked):
+            resp = await self.client.get(url)
+            self.assertEqual(resp.status, 200)
+            self.assertEqual(resp.headers["Cache-Control"], "no-store")
+            await resp.read()
+        self.assertEqual(self.api._auto_thumbnails().failure_count, 0)
+
+        resp = await self.client.get(url)
+        self.assertEqual(resp.status, 200)
+        self.assertTrue((self.output_dir / "thumbnails" / "a_thumb.png").is_file())
+
+    async def test_no_source_is_a_plain_uncacheable_404(self):
+        resp = await self.client.get(f"{SERVE}thumbnails/nope_thumb.png?root=0")
+        self.assertEqual(resp.status, 404)
+        self.assertEqual(resp.headers["Cache-Control"], "no-store")
+        text = await resp.text()
+        self.assertNotIn(str(self.output_dir), text)
+        self.assertFalse((self.output_dir / "thumbnails").exists())
+
+    async def test_traversal_is_refused_before_any_filesystem_access(self):
+        with patch.object(auto, "_thumbnail_source") as source:
+            for path in (
+                "thumbnails/../../etc/passwd_thumb.png",
+                "thumbnails/..%2F..%2Fx_thumb.png",
+            ):
+                resp = await self.client.get(f"{SERVE}{path}?root=0")
+                self.assertIn(resp.status, (403, 404))
+            source.assert_not_called()
+
+    async def test_symlinked_source_is_never_read(self):
+        outside = Path(self._output_tmp.name) / "outside.png"
+        make_png(outside, size=(64, 64))
+        self._make_symlink(outside, self.output_dir / "link.png")
+        resp = await self.client.get(f"{SERVE}thumbnails/link_thumb.png?root=0")
+        self.assertEqual(resp.status, 404)
+        self.assertFalse((self.output_dir / "thumbnails").exists())
+
+    async def test_thumbnails_folder_linked_out_of_the_root_is_never_written(self):
+        make_png(self.output_dir / "a.png", size=(64, 64))
+        elsewhere = Path(self._output_tmp.name) / "elsewhere"
+        elsewhere.mkdir()
+        self._make_symlink(elsewhere, self.output_dir / "thumbnails")
+        resp = await self.client.get(f"{SERVE}thumbnails/a_thumb.png?root=0")
+        self.assertEqual(resp.status, 200)
+        self.assertEqual(resp.headers["Cache-Control"], "no-store")
+        self.assertEqual(os.listdir(elsewhere), [])
+
+    async def test_restore_hold_serves_the_original_and_writes_nothing(self):
+        from py.api.jobs import RESTORE_JOB_KIND
+
+        make_png(self.output_dir / "a.png", size=(64, 64))
+        url = _entry(self.api, self.output_dir, "a.png")["thumbnail_url"]
+        self._hold(RESTORE_JOB_KIND)
+        resp = await self.client.get(url)
+        self.assertEqual(resp.status, 200)
+        self.assertEqual(resp.headers["Cache-Control"], "no-store")
+        self.assertEqual(await resp.read(), (self.output_dir / "a.png").read_bytes())
+        self.assertFalse((self.output_dir / "thumbnails").exists())
+        self.assertEqual(self.api._auto_thumbnails().failure_count, 0)
+
+    async def test_duplicate_removal_and_source_change_hold_writes(self):
+        from py.api.jobs import DUPLICATE_CLEANUP_JOB_KIND
+        from py.api.settings_routes import SOURCE_CHANGE_JOB_KIND
+
+        make_png(self.output_dir / "a.png", size=(64, 64))
+        for kind in (DUPLICATE_CLEANUP_JOB_KIND, SOURCE_CHANGE_JOB_KIND):
+            with self.subTest(kind=kind):
+                job = self._hold(kind, title=kind)
+                self.assertIs(self.api._auto_thumbnail_blocker(), job)
+                job.finish("finished")
+        self.assertIsNone(self.api._auto_thumbnail_blocker())
+
+    async def test_scan_and_thumbnail_jobs_do_not_hold_writes(self):
+        # A scan writes records only and the thumbnail job writes the same
+        # previews atomically, so a view during either still gets its preview.
+        holds = (
+            ("scan", "Scan for metadata", "records"),
+            ("image-tags", "Add tags", "records"),
+            (THUMBNAILS_JOB_KIND, "Thumbnail generation", "guarded"),
+        )
+        for index, (kind, title, admission) in enumerate(holds):
+            with self.subTest(kind=kind):
+                name = f"a{index}.png"
+                make_png(self.output_dir / name, size=(64, 64))
+                job = self._hold(kind, title=title, admission=admission)
+                resp = await self.client.get(
+                    _entry(self.api, self.output_dir, name)["thumbnail_url"]
+                )
+                self.assertEqual(resp.status, 200)
+                thumb = self.output_dir / "thumbnails" / f"a{index}_thumb.png"
+                self.assertTrue(thumb.is_file())
+                job.finish("finished")
+
+    async def test_legacy_clear_holds_writes(self):
+        make_png(self.output_dir / "a.png", size=(64, 64))
+        self._hold(
+            THUMBNAILS_JOB_KIND,
+            title=images_module.THUMBNAIL_CLEAR_TITLE,
+            admission="guarded",
+        )
+        resp = await self.client.get(
+            _entry(self.api, self.output_dir, "a.png")["thumbnail_url"]
+        )
+        self.assertEqual(resp.status, 200)
+        self.assertFalse((self.output_dir / "thumbnails").exists())
+
+
+class TestThumbnailOnCapture(AutoThumbnailCase):
+    async def test_monitor_output_is_queued_and_generated_in_the_background(self):
+        path = make_png(self.output_dir / "run" / "ComfyUI_0001.png", size=(640, 480))
+        outcome = self.api.queue_capture_thumbnail(str(path))
+        self.assertEqual(outcome, "queued")
+        await self._idle()
+        thumb = self.output_dir / "thumbnails" / "run" / "ComfyUI_0001_thumb.png"
+        self.assertTrue(thumb.is_file())
+        self.assertIn(
+            "v=",
+            _entry(self.api, self.output_dir, "run/ComfyUI_0001.png")["thumbnail_url"],
+        )
+
+    async def test_duplicates_coalesce_and_fresh_previews_are_skipped(self):
+        path = make_png(self.output_dir / "a.png", size=(64, 64))
+        calls = self._count_writes()
+        entered, release, _state = self._gate_writes()
+        self.assertEqual(self.api.queue_capture_thumbnail(str(path)), "queued")
+        await self._wait(entered)
+        # While it is being written a second event for it is not queued again.
+        other = make_png(self.output_dir / "b.png", size=(64, 64))
+        self.assertEqual(self.api.queue_capture_thumbnail(str(other)), "queued")
+        self.assertEqual(self.api.queue_capture_thumbnail(str(other)), "skipped")
+        release.set()
+        await self._idle()
+        self.assertEqual(sorted(calls), ["a.png", "b.png"])
+        self.assertEqual(self.api.queue_capture_thumbnail(str(path)), "fresh")
+
+    async def test_files_outside_the_roots_and_previews_are_ignored(self):
+        outside = make_png(Path(self._output_tmp.name) / "x.png")
+        preview = make_png(self.output_dir / "thumbnails" / "y_thumb.png")
+        text = self.output_dir / "notes.txt"
+        text.write_text("x")
+        for path in (outside, preview, text, self.output_dir / "missing.png"):
+            self.assertEqual(self.api.queue_capture_thumbnail(str(path)), "skipped")
+
+    async def test_worker_uses_the_worker_threads_setting(self):
+        from py.config import PromptManagerConfig
+
+        original = PromptManagerConfig.WORKER_THREADS
+        PromptManagerConfig.WORKER_THREADS = 2
+        self.addCleanup(setattr, PromptManagerConfig, "WORKER_THREADS", original)
+        batches = []
+        real_map = auto.map_parallel
+
+        def recording(func, items, workers):
+            batches.append((len(items), workers))
+            return real_map(func, items, workers)
+
+        patcher = patch.object(auto, "map_parallel", recording)
+        patcher.start()
+        self.addCleanup(patcher.stop)
+        entered, release, state = self._gate_writes()
+        paths = [
+            make_png(self.output_dir / f"w{i}.png", size=(32, 32)) for i in range(5)
+        ]
+        self.api.queue_capture_thumbnail(str(paths[0]))
+        await self._wait(entered)  # the worker holds a batch of one
+        for path in paths[1:]:
+            self.api.queue_capture_thumbnail(str(path))
+        release.set()
+        await self._idle()
+        self.assertEqual(batches, [(1, 1), (2, 2), (2, 2)])
+        self.assertLessEqual(state["peak"], 2)
+        for i in range(5):
+            self.assertTrue(
+                (self.output_dir / "thumbnails" / f"w{i}_thumb.png").is_file()
+            )
+
+    async def test_worker_survives_a_failing_write(self):
+        bad = self.output_dir / "bad.png"
+        bad.write_bytes(b"broken")
+        good = make_png(self.output_dir / "good.png", size=(64, 64))
+        self.api.queue_capture_thumbnail(str(bad))
+        await self._idle()
+        self.api.queue_capture_thumbnail(str(good))
+        await self._idle()
+        self.assertTrue((self.output_dir / "thumbnails" / "good_thumb.png").is_file())
+        self.assertEqual(self.api._auto_thumbnails().failure_count, 1)
+
+    async def test_worker_writes_nothing_during_a_restore(self):
+        from py.api.jobs import RESTORE_JOB_KIND
+
+        path = make_png(self.output_dir / "a.png", size=(64, 64))
+        self._hold(RESTORE_JOB_KIND)
+        self.api.queue_capture_thumbnail(str(path))
+        await self._idle()
+        self.assertFalse((self.output_dir / "thumbnails").exists())
+
+    def test_queue_is_bounded(self):
+        thumbnailer = auto.AutoThumbnailer(
+            lambda *a: {"action": "generated"},
+            lambda: True,
+            self.api.logger,
+            max_queued=2,
+            start_worker=False,
+        )
+        paths = [Path(f"/x/{i}.png") for i in range(3)]
+        added = [thumbnailer.enqueue(p, p.with_suffix(".t.png"), False) for p in paths]
+        self.assertEqual(added, [True, True, False])
+        thumbnailer.close()
+
+
+class TestMonitorQueuesThumbnails(AutoThumbnailCase):
+    def _handler(self):
+        from utils.image_monitor import ImageGenerationHandler
+
+        handler = ImageGenerationHandler(self.api.db, unittest.mock.MagicMock())
+        handler.processing_delay = 0
+        handler.settle_interval = 0
+        return handler
+
+    def test_settled_output_is_handed_to_the_thumbnail_queue(self):
+        path = make_png(self.output_dir / "n.png", size=(32, 32))
+        handler = self._handler()
+        calls = []
+        handler.thumbnail_queue = calls.append
+        with patch.object(handler, "process_new_image"):
+            handler._pending[str(path)] = (None, 0.0)
+            handler._process_pending(str(path))
+        self.assertEqual(calls, [str(path)])
+
+    def test_a_failing_queue_never_breaks_the_monitor(self):
+        path = make_png(self.output_dir / "n.png", size=(32, 32))
+        handler = self._handler()
+
+        def broken(_path):
+            raise RuntimeError("boom")
+
+        handler.thumbnail_queue = broken
+        with patch.object(handler, "process_new_image"):
+            handler._pending[str(path)] = (None, 0.0)
+            with self.assertLogs("prompt_manager.image_monitor", "WARNING") as logs:
+                handler._process_pending(str(path))  # no exception
+        self.assertIn("could not be queued", "\n".join(logs.output))
+        self.assertEqual(handler.pending_count(), 0)
+
+    def test_monitor_wiring_reaches_new_and_existing_handlers(self):
+        from utils.image_monitor import ImageMonitor
+        from utils.prompt_tracker import PromptTracker
+
+        monitor = ImageMonitor(self.api.db, PromptTracker(self.api.db))
+        queue_fn = self.api.queue_capture_thumbnail
+        monitor.set_thumbnail_queue(queue_fn)
+        monitor.start_monitoring([str(self.output_dir)])
+        try:
+            self.assertEqual(monitor.handler.thumbnail_queue, queue_fn)
+            monitor.set_thumbnail_queue(None)
+            self.assertIsNone(monitor.handler.thumbnail_queue)
+        finally:
+            monitor.stop_monitoring()
+
+
+class TestScanHandsOffThumbnails(AutoThumbnailCase):
+    def _scan_job(self, added):
+        from py.api.jobs import Job, SCAN_GROUP_NAMES
+
+        job = Job(
+            "scan", "Scan for metadata", admission="records", groups=SCAN_GROUP_NAMES
+        )
+        job.counters["added"] = added
+        job.finish("finished")
+        return job
+
+    async def test_many_missing_previews_start_the_visible_thumbnail_job(self):
+        for i in range(3):
+            make_png(self.output_dir / f"s{i}.png", size=(32, 32))
+        with patch.object(auto, "SCAN_THUMBNAIL_JOB_THRESHOLD", 3):
+            outcome = await self.api._thumbnails_after_scan(self._scan_job(3))
+        self.assertEqual(outcome, "job")
+        job = self.api._jobs.active(THUMBNAILS_JOB_KIND)
+        self.assertIsNotNone(job)
+        self.assertTrue(job.cancellable)
+        self.assertEqual(job.scope["mode"], "missing")
+        await asyncio.wait_for(job.task, WAIT)
+        self.assertEqual(job.counters["generated"], 3)
+
+    async def test_few_missing_previews_go_through_the_worker(self):
+        for i in range(2):
+            make_png(self.output_dir / f"s{i}.png", size=(32, 32))
+        with patch.object(auto, "SCAN_THUMBNAIL_JOB_THRESHOLD", 3):
+            outcome = await self.api._thumbnails_after_scan(self._scan_job(2))
+        self.assertEqual(outcome, "queued")
+        self.assertIsNone(self.api._jobs.active(THUMBNAILS_JOB_KIND))
+        await self._idle()
+        for i in range(2):
+            self.assertTrue(
+                (self.output_dir / "thumbnails" / f"s{i}_thumb.png").is_file()
+            )
+
+    async def test_a_scan_that_added_nothing_does_nothing(self):
+        make_png(self.output_dir / "s.png", size=(32, 32))
+        self.assertIsNone(await self.api._thumbnails_after_scan(self._scan_job(0)))
+        self.assertFalse((self.output_dir / "thumbnails").exists())
+
+    async def _run_scan(self, final):
+        calls = []
+
+        async def events(job, output_dirs=None):
+            job.counters["added"] = 1
+            yield {"type": final}
+
+        async def hand_off(job, output_dirs=None):
+            calls.append((job.state, output_dirs))
+
+        self.api._scan_images_events = events
+        self.api._hand_off_scan_thumbnails = hand_off
+        job = self.api._start_scan_job("new")
+        await asyncio.wait_for(job.task, WAIT)
+        return calls
+
+    async def test_a_finished_scan_hands_its_previews_on(self):
+        self.assertEqual(await self._run_scan("complete"), [("finished", None)])
+
+    async def test_a_cancelled_or_failed_scan_does_not(self):
+        self.assertEqual(await self._run_scan("cancelled"), [])
+        self.assertEqual(await self._run_scan("error"), [])
+
+    async def test_hand_off_failures_are_logged_not_raised(self):
+        async def broken(job, output_dirs=None):
+            raise OSError("walk failed")
+
+        self.api._thumbnails_after_scan = broken
+        with self.assertLogs("prompt_manager.api", "WARNING"):
+            self.assertIsNone(
+                await self.api._hand_off_scan_thumbnails(self._scan_job(3))
+            )
+
+    async def test_a_scoped_scan_counts_only_its_sources(self):
+        other = self.output_dir.parent / "other"
+        make_png(other / "o.png", size=(32, 32))
+        make_png(self.output_dir / "s.png", size=(32, 32))
+        job = self._scan_job(1)
+        job.scope["roots"] = [1]
+        outcome = await self.api._thumbnails_after_scan(job, [self.output_dir, other])
+        self.assertEqual(outcome, "queued")
+        await self._idle()
+        self.assertTrue((other / "thumbnails" / "o_thumb.png").is_file())
+        self.assertFalse((self.output_dir / "thumbnails").exists())
+
+    async def test_nothing_missing_is_none(self):
+        make_png(self.output_dir / "s.png", size=(32, 32))
+        make_png(self.output_dir / "thumbnails" / "s_thumb.png")
+        self.assertEqual(
+            await self.api._thumbnails_after_scan(self._scan_job(1)), "none"
+        )
+
+    async def test_a_busy_registry_is_reported_not_raised(self):
+        for i in range(3):
+            make_png(self.output_dir / f"s{i}.png", size=(32, 32))
+        self._hold("restore")
+        with patch.object(auto, "SCAN_THUMBNAIL_JOB_THRESHOLD", 3):
+            outcome = await self.api._thumbnails_after_scan(self._scan_job(3))
+        self.assertEqual(outcome, "busy")
+
+
+if __name__ == "__main__":
+    unittest.main()

utils/image_monitor.py

@@ -79,6 +79,10 @@ class ImageGenerationHandler(FileSystemEventHandler):
         # ``record_review_file``, wired at start-up): a linked output is
         # never queued and its pending items leave the queue. None: no pass.
         self.review_recorder: Optional[Callable[[str], str]] = None
+        # Queues a settled output's preview (the API's
+        # ``queue_capture_thumbnail``, wired at start-up; #249). It only
+        # queues: the preview is written on its own worker. None: no queue.
+        self.thumbnail_queue: Optional[Callable[[str], str]] = None
 
         # Read from GalleryConfig if available, otherwise use defaults
         try:
@@ -220,10 +224,30 @@ class ImageGenerationHandler(FileSystemEventHandler):
                     )
                 return
             self.process_new_image(image_path, prompt_snapshot=prompt_snapshot)
+            self._queue_thumbnail(image_path)
         finally:
             with self._pending_lock:
                 self._pending.pop(image_path, None)
 
+    def _queue_thumbnail(self, image_path: str) -> None:
+        """Queue a settled output's preview, best effort (linked or not).
+
+        A failure is logged, never raised: the preview is then made on its
+        first view.
+        """
+        queue_preview = getattr(self, "thumbnail_queue", None)
+        if queue_preview is None:
+            return
+        try:
+            outcome = queue_preview(image_path)
+        except Exception:
+            self.logger.warning(
+                f"Preview of {os.path.basename(image_path)} could not be queued",
+                exc_info=True,
+            )
+            return
+        self.logger.debug(f"Preview {outcome}: {os.path.basename(image_path)}")
+
     def wait_until_settled(self, image_path: str, stop=None) -> bool:
         """True once the file's size is non-zero and unchanged across two reads.
 
@@ -636,6 +660,16 @@ class ImageMonitor:
         self.monitored_directories = []
         self.logger = get_logger("prompt_manager.image_monitor")
         self.review_recorder: Optional[Callable[[str], str]] = None
+        self.thumbnail_queue: Optional[Callable[[str], str]] = None
+
+    def set_thumbnail_queue(
+        self, queue_preview: Optional[Callable[[str], str]]
+    ) -> None:
+        """Queue every settled output's preview through ``queue_preview``
+        (None stops it), on the running handler and every later one."""
+        self.thumbnail_queue = queue_preview
+        if self.handler is not None:
+            self.handler.thumbnail_queue = queue_preview
 
     def set_review_recorder(self, recorder: Optional[Callable[[str], str]]) -> None:
         """Record ambiguous outputs through ``recorder`` (None stops recording).
@@ -703,6 +737,7 @@ class ImageMonitor:
         # leaves the monitor stuck "already running" with nothing to stop.
         handler = ImageGenerationHandler(self.db_manager, self.prompt_tracker)
         handler.review_recorder = self.review_recorder
+        handler.thumbnail_queue = self.thumbnail_queue
         observer = Observer()
         scheduled = []
         for output_dir in valid_directories:

web/js/workspace/file-inspector.mjs

@@ -318,10 +318,13 @@ export function mountFileInspector(container, {api, sender, parentScope, variant
         });
         return button;
     }
-    /** The concept's plain cover; the card image opens the viewer (D19). Video keeps its own element. */
+    /**
+     * The concept's plain cover; the card image opens the viewer (D19). #249: it is the file's preview (a video's
+     * is a JPEG frame) whenever the listing gave one; without one the original, and a video keeps its own element.
+     */
     function cover(record) {
-        const {file} = record, video = file?.media_type === 'video';
-        const node = el(video ? 'video' : 'img', undefined, 'detail-cover'); node.src = record.download.url;
+        const {file} = record, preview = record.preview?.url, video = !preview && file?.media_type === 'video';
+        const node = el(video ? 'video' : 'img', undefined, 'detail-cover'); node.src = preview || record.download.url;
         if (video) {
             node.muted = true; node.playsInline = true; node.preload = 'metadata';
             contentScope.cleanup(() => {node.pause(); node.removeAttribute('src'); node.load();});

web/js/workspace/history-media.mjs

@@ -64,7 +64,10 @@ export function toHistoryInspectionRecord(result, target, extractFacts) {
     const path = [target.reference.subfolder, target.reference.filename].filter(Boolean).join('/');
     const identity = {kind:'reported-location', key:target.location_key, path:target.reference.path_style === 'windows' ? path.replaceAll('\\', '/') : path,
         occurrence:{...target.occurrence}, history_epoch:target.history_epoch, current_scope:target.current_scope, reference:{...target.reference}};
-    return projectInspectionPayload(result, identity, {url:urlFor(target, 'url', 'media'), filename:target.filename}, extractFacts);
+    const record = projectInspectionPayload(result, identity, {url:urlFor(target, 'url', 'media'), filename:target.filename}, extractFacts);
+    // #249: the inspector's cover is the recorded output's preview (made by the server on first view).
+    record.preview = {url:urlFor(target, 'thumbnail_url', 'thumbnail')};
+    return record;
 }
 export const reportedLocationAdapter = Object.freeze({
     validate,

web/js/workspace/inspection-state.mjs

@@ -68,6 +68,39 @@ function downloadFor(image, required) {
     }
 }
 
+// The server's video extensions (py/api/inspection_service.py VIDEO_EXTENSIONS): their preview is a JPEG.
+const VIDEO_EXTENSIONS = new Set(['.mp4', '.webm', '.avi', '.mov', '.mkv', '.m4v', '.wmv']);
+
+/** `thumbnails/<path>_thumb<ext>` for a listed file, as the server names its preview (images.media_thumbnail_rel). */
+function previewPath(path) {
+    const slash = path.lastIndexOf('/'), dot = path.lastIndexOf('.');
+    const cut = dot > slash ? dot : path.length, extension = path.slice(cut).toLowerCase();
+    return `thumbnails/${path.slice(0, cut)}_thumb${VIDEO_EXTENSIONS.has(extension) ? '.jpg' : extension}`;
+}
+
+/**
+ * #249: the listing's thumbnail of `image` as `{url}` for the inspector's cover, or null. Only the same-origin
+ * serve URL of this file's own preview under the same root and source scope is used; anything else keeps the
+ * original (a missing preview is made by the server on first view).
+ */
+function previewFor(image) {
+    try {
+        if (typeof image.thumbnail_url !== 'string' || !image.thumbnail_url) return null;
+        const location = globalThis.location || globalThis.window?.location;
+        const reference = new URL(location?.href || 'http://localhost/');
+        const resolveUrl = globalThis.window?.ApiClient?.resolveUrl || globalThis.ApiClient?.resolveUrl || (url => url);
+        const value = resolveUrl(image.thumbnail_url), url = new URL(value, reference);
+        const expected = new URL(resolveUrl(`/prompt_manager/images/serve/${previewPath(image.path).split('/').map(encodeURIComponent).join('/')}`), reference);
+        const scope = descriptorScope(image);
+        if (url.origin !== reference.origin || url.username || url.password || url.hash || url.pathname !== expected.pathname ||
+            url.searchParams.getAll('root').length !== 1 || url.searchParams.get('root') !== String(image.root_index) ||
+            url.searchParams.getAll('root_scope').length > 1 || (scope !== null && url.searchParams.get('root_scope') !== scope)) return null;
+        return { url: value };
+    } catch {
+        return null;
+    }
+}
+
 export function originalMediaDownload(image) {
     validateIdentity(image);
     return downloadFor(image, true);
@@ -284,16 +317,18 @@ export function toInspectionRecord(result, image, extractFacts = globalThis.Comf
     if (!legacy && result.inspection_version !== 1) throw new TypeError('Unsupported inspection version');
     if (!legacy && result.success !== true) throw new TypeError('Invalid inspection success status');
     if ((!legacy || result.image !== undefined) && (!object(result.image) || result.image.root_index !== image.root_index || result.image.path !== image.path || result.image.filename !== image.path.split('/').at(-1))) throw new TypeError('Mismatched inspection identity');
-    return projectInspectionPayload(result,
+    const record = projectInspectionPayload(result,
         { kind: 'rooted-file', key: imageKey(image), root_index: image.root_index, path: image.path, root_scope: scope },
         downloadFor(image, !legacy || scope !== null), extractFacts, legacy, parser);
+    record.preview = record.download ? previewFor(image) : null;
+    return record;
 }
 
 /** Shared bounded payload projection; adapters validate identity before entering here. */
 export function projectInspectionPayload(result, identity, download, extractFacts = globalThis.ComfyMetadata?.extractGenerationFacts || globalThis.window?.ComfyMetadata?.extractGenerationFacts, legacy = false, parser = comfyMetadata()) {
     const record = {
         identity, file: null, prompts: promptsFor(result.prompts), library: null, metadata: null,
-        generation: unavailableFacts(), historyTarget: null, download, imageTags: null,
+        generation: unavailableFacts(), historyTarget: null, download, imageTags: null, preview: null,
     };
     if (legacy) return record;
     record.imageTags = imageTagsFor(result.image_tags);

web/js/workspace/system-overview.mjs

@@ -88,7 +88,9 @@ export function thumbnailTitle(status, sources) {
     const missing = status?.pending;
     if (!Number.isSafeInteger(missing) || sources?.effective === 0) return 'Thumbnail status unavailable';
     if (missing === 0) return 'No missing thumbnails';
-    return `${number(missing)} missing ${noun(missing, 'thumbnail')}`;
+    // #249: files whose automatic preview could not be made this session (they show the original).
+    const failed = Number.isSafeInteger(status.failed) && status.failed > 0 ? ` · ${number(status.failed)} failed` : '';
+    return `${number(missing)} missing ${noun(missing, 'thumbnail')}${failed}`;
 }
 
 const SOURCES_SHARED = 'Monitoring and enrichment share one source list.';