fix(thumbnails): answer shared flights with the outcome only and restart a failed worker

The history and serve routes share a flight key, so a flight now answers
only the ensure outcome whichever route started it; the serve route
derives the preview path and, for a fallback, looks its source up again
on the view pool. A view joining a history flight no longer gets a 500
or an image answered with the video placeholder.

The capture worker clears its running state however its loop ends, and a
worker that cannot start leaves the file queued for the next enqueue.

TESTING.md

@@ -1932,8 +1932,8 @@ pending count (`GET /prompt_manager/review/items?limit=1`, its `total`).
   unavailable file's empty media button keeps the concept's 1.33 media area, so its
   card lines up with the others. A long file name in the panel's `p.small` wraps inside
   the panel (`overflow-wrap: anywhere`), where the concept's short sample name never
-  overflows. A queue card shows the Images listing's thumbnail when the server knows one
-  (`thumbnail_url`, the same URL the Images page uses), else the original; a re-read
+  overflows. A queue card shows the Images listing's thumbnail (`thumbnail_url`, the
+  same URL the Images page uses; a preview not made yet is made on first view); a re-read
   rebuilds a card whose media, target or availability changed (a root came or went), so
   Open and Tag follow it. When an unanswered decision's item leaves the queue in a
   re-read, the decision was recorded: the page finishes as a recorded decision does (its
@@ -2230,6 +2230,59 @@ through the Activity store, which toasts "Thumbnail generation started."; the jo
 in Activity. A start refused because another job holds the slot toasts the D57 busy
 link toast and stays. The legacy gallery's thumbnail tools are unchanged.
 
+Automatic previews (#249, `py/api/thumbnail_auto.py`). The approved mock has no
+repeated global thumbnail prompt (`docs/concepts/workspace/README.md`), so the
+workspace never asks; previews are a cache the server keeps:
+
+- Every listing gives an image or video a thumbnail URL (`media_thumbnail_url`: Images,
+  saved views and favorites, Library cards and galleries, Needs review, duplicate
+  groups). An existing preview's URL keeps its `v=<mtime>`; a missing one is listed
+  without `v=`.
+- `GET /images/serve/thumbnails/<rel>_thumb<ext>` for a missing preview is refused as
+  text first (traversal, a backslash, a drive or stream colon, empty or dot segments, so
+  `D:/x`, `\\server\share` and a leading `/` never reach a joined path), then admitted
+  on the event loop (`_PreviewGate`): 3 previews are written at once on the API's own
+  `pm-preview` pool (never the default executor), one flight runs per file (the history
+  route shares the flight key and the admission), and at most 64 requests wait; one
+  more gets its fallback at once without listing a folder. Inside the admission the
+  preview maps back to its media file under the same root (exact names, then one folder
+  listing for a mixed-case extension; a regular file inside the root, never through a
+  link) and is made with the job's defaults (512 px, medium). When it cannot be made an
+  image's original is served and a video's placeholder frame (a video is never sent to
+  an `<img>`), both `Cache-Control: no-store`; cards and the inspector cover also swap a
+  video back to its own element when its preview image fails. Only a file that cannot
+  be decoded is remembered for the session (path, size, mtime; retried once it changes)
+  and counted as `failed` in `/images/thumbnails/status`, which Overview's row adds
+  ("N missing thumbnails · M failed"); I/O errors and a preview another program holds
+  open (Windows) are not. A preview folder (or any link segment of it) resolving out of
+  the root is never written. Media inside a nested `thumbnails` folder is listed with no
+  thumbnail and keeps its original. The listing's `thumbnail_ready` says whether the
+  preview file exists (the legacy gallery's dot and "Fast").
+- The video decoder falls through OpenCV, ffmpeg (a tracked process, killed on
+  shutdown) and the placeholder frame: an OpenCV that cannot open or read a clip (AV1)
+  no longer ends the run.
+- The image monitor hands every settled output to `queue_capture_thumbnail`, which
+  queues it for one background worker: each wake takes at most Worker threads files and
+  writes them on that many threads; a queued file is not queued twice; the queue holds
+  10,000 and a file left out is made on view.
+- A finished scan that added files counts the missing previews of the sources it
+  walked: 200 or more start the thumbnail job (missing only, visible and cancellable in
+  Activity, never silent bulk work); fewer go through the worker.
+- Holds: automatic previews write only `thumbnails/` files and no records, so they run
+  beside a scan, image tagging, the thumbnail job and the Needs review recording (S5
+  M7). They wait (a view gets the original, a queued file is left for its view) while a
+  restore, a duplicate removal, a sources change, the gallery's thumbnail clear or an
+  `exclusive` job runs (`_auto_thumbnail_blocker`), checked when a write starts (after
+  its slot) and again just before its rename.
+- Shutdown is bounded and logged: the worker stops at a file boundary, ffmpeg is
+  killed, an idle worker leaves (the next output starts a fresh one), the view pool
+  shuts down, and after the app's cleanup a late monitor event
+  starts nothing. A package reload stops the workers the previous import left.
+- Cards and strips load previews only: Images in every layout and size, Library cards,
+  Needs review cards and the inspector's cover (a video's cover is its preview frame).
+  The media viewer's main media and downloads load the original
+  (`workspace-thumbnails.spec.mjs` records every media request; `tests/test_thumbnail_auto.py`).
+
 Overview (`system-overview.mjs`) reads `GET /prompt_manager/system/overview` once per
 visit through `latestOnly()` on the section scope; a late answer after the section left
 paints nothing. Until it answers, the section shows "Loading the overview…"; a failure
@@ -2277,7 +2330,8 @@ Recorded deviations (plan D60–D67, D79–D81 renumbered D92–D99 and D111–D
   "Backup status" reads "No retained backup record." until verified backups exist (S8).
   The thumbnail row is "Checking thumbnails…" until its own read answers, then "No
   missing thumbnails", "N missing thumbnails" (the pending count of
-  `/images/thumbnails/status`) or "Thumbnail status unavailable"; Review thumbnails
+  `/images/thumbnails/status`, followed by " · M failed" when automatic previews
+  failed this session) or "Thumbnail status unavailable"; Review thumbnails
   opens the thumbnail dialog (D99). "One display configuration" says "Grid
   and density settings are shared across views." (no paging). The sources row is
   "N output sources" with "Monitoring and enrichment share one source list.", prefixed

__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(final=True)
 
             _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
@@ -48,6 +49,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 .workflow_storage import WorkflowStorageRoutesMixin
@@ -291,6 +293,7 @@ class PromptManagerAPI(
     JobRoutesMixin,
     ImageTagJobsMixin,
     ThumbnailJobsMixin,
+    AutoThumbnailMixin,
     SystemRoutesMixin,
     MaintenanceRoutesMixin,
     WorkflowStorageRoutesMixin,

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",
+                        **media_options,
                     )
                 _check_current(scope, self, fence, epoch, generation)
                 return prepared
@@ -761,25 +764,44 @@ class HistoryRoutesMixin:
                     _check_current(scope, self, fence, epoch, generation)
                 raise
 
-        pending = asyncio.create_task(self._run_in_executor(project))
-        try:
-            prepared = await asyncio.shield(pending)
-        except asyncio.CancelledError:
-            pending.add_done_callback(_close_prepared_result)
-            raise
-        except (
-            HistoryReadError,
-            HistoryEpochChanged,
-            sqlite3.Error,
-            OSError,
-            ValueError,
-        ):
-            raise
-        except Exception as error:
-            self.logger.warning(
-                "Current history media unavailable (%s)", type(error).__name__
-            )
-            raise _current_unavailable("temporary_unavailable") from None
+        async def execute():
+            pending = asyncio.create_task(self._run_in_executor(project))
+            try:
+                return await asyncio.shield(pending)
+            except asyncio.CancelledError:
+                pending.add_done_callback(_close_prepared_result)
+                raise
+            except (
+                HistoryReadError,
+                HistoryEpochChanged,
+                sqlite3.Error,
+                OSError,
+                ValueError,
+            ):
+                raise
+            except Exception as error:
+                self.logger.warning(
+                    "Current history media unavailable (%s)", type(error).__name__
+                )
+                raise _current_unavailable("temporary_unavailable") from None
+
+        # A missing preview (#249) is noted on the first read, made through
+        # the serve route's admission and single flight, then read again. A
+        # video whose preview cannot be made is the placeholder frame.
+        needed = []
+        media_options = {}
+        if kind == "thumbnail":
+            media_options = {
+                "ensure_preview": lambda *preview: needed.append(preview),
+                "placeholder": self._video_placeholder_bytes(),
+            }
+        prepared = await execute()
+        if needed:
+            if not isinstance(prepared, bytes):
+                prepared.close()
+            await self._admit_preview(*needed[0])
+            media_options["ensure_preview"] = None
+            prepared = await execute()
         if isinstance(prepared, bytes):
             return _success_response(prepared)
         return await _current_media_response(request, prepared)

py/api/history_current.py

@@ -5,6 +5,7 @@ from dataclasses import dataclass
 import errno
 import hashlib
 import hmac
+import io
 import json
 import mimetypes
 import os
@@ -309,17 +310,35 @@ class _OpenedCurrentMedia:
         self.handle.close()
 
 
+def _preview_writable(root: Path, preview: Path) -> bool:
+    """A preview may be written: ``thumbnails`` and every link segment of the
+    preview's folder resolve inside *root* (the serve route's rule, #249)."""
+    folder = root / "thumbnails"
+    return _canonical(folder).is_relative_to(_canonical(root)) and _canonical(
+        preview
+    ).is_relative_to(_canonical(folder))
+
+
 def _prepare_current_media(
-    resolved: ResolvedCurrentCandidate, scope: CurrentOutputScope, *, thumbnail: bool
+    resolved: ResolvedCurrentCandidate,
+    scope: CurrentOutputScope,
+    *,
+    thumbnail: bool,
+    ensure_preview=None,
+    placeholder: bytes | None = None,
 ) -> _OpenedCurrentMedia:
+    """Open the current file, or its preview for ``thumbnail``.
+
+    A missing preview that may be written is handed to
+    ``ensure_preview(src, dst, is_video)`` (the route makes it and reads
+    again, #249). Without one an image's original is opened and a video is
+    the ``placeholder`` JPEG, never the video inside an ``<img>``.
+    """
     target = resolved.canonical_target
     if thumbnail:
         relative = target.relative_to(resolved.gallery_root)
-        extension = (
-            ".jpg"
-            if target.suffix.lower() in VIDEO_EXTENSIONS
-            else target.suffix.lower()
-        )
+        is_video = target.suffix.lower() in VIDEO_EXTENSIONS
+        extension = ".jpg" if is_video else target.suffix.lower()
         preview = (
             resolved.gallery_root
             / "thumbnails"
@@ -328,7 +347,14 @@ def _prepare_current_media(
         try:
             preview.stat()
         except FileNotFoundError:
-            pass
+            if ensure_preview is not None and _preview_writable(
+                resolved.gallery_root, preview
+            ):
+                ensure_preview(target, preview, is_video)
+            if is_video and placeholder is not None:
+                return _OpenedCurrentMedia(
+                    io.BytesIO(placeholder), len(placeholder), "image/jpeg"
+                )
         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):
@@ -261,13 +263,32 @@ def _temp_sibling(dst):
     return dst.with_name(f".{dst.stem}.part-{token}{dst.suffix}")
 
 
-def _atomic_save(img, dst, fmt=None, **save_kwargs):
+class PreviewWriteHeld(Exception):
+    """A preview write stopped before its rename: ``guard()`` said no (#249)."""
+
+
+# The result error of a write held before its rename (never remembered).
+PREVIEW_HELD = "Preview writes are held while another job runs"
+
+
+def _replace_unless_held(tmp, dst, guard=None):
+    """Rename *tmp* over *dst*, unless ``guard()`` is false (PreviewWriteHeld)."""
+    if guard is not None and not guard():
+        raise PreviewWriteHeld(PREVIEW_HELD)
+    os.replace(tmp, dst)
+
+
+def _atomic_save(img, dst, fmt=None, *, guard=None, **save_kwargs):
     """Save a PIL image to *dst* without a partially written file ever being
-    visible at that path: write a temp sibling, then rename over it."""
+    visible at that path: write a temp sibling, then rename over it.
+
+    ``guard()`` (automatic previews) is asked again just before the rename;
+    false leaves *dst* untouched and raises PreviewWriteHeld.
+    """
     tmp = _temp_sibling(dst)
     try:
         img.save(tmp, fmt, **save_kwargs)
-        os.replace(tmp, dst)
+        _replace_unless_held(tmp, dst, guard)
     finally:
         if tmp.exists():
             tmp.unlink()
@@ -305,11 +326,11 @@ def _ffmpeg_qscale(encoding):
     return str(_FFMPEG_QSCALE.get(encoding["quality"], 4))
 
 
-def _write_image_thumbnail(src, dst, thumbnail_size, encoding=None):
+def _write_image_thumbnail(src, dst, thumbnail_size, encoding=None, guard=None):
     """Resize *src* into *dst* (blocking PIL work), atomically.
 
     ``encoding`` is one of THUMBNAIL_ENCODINGS (medium when None); a PNG
-    preview keeps only its ``optimize`` flag.
+    preview keeps only its ``optimize`` flag. ``guard`` see _atomic_save.
     """
     encoding = encoding or DEFAULT_THUMBNAIL_ENCODING
     with Image.open(src) as img:
@@ -319,7 +340,50 @@ def _write_image_thumbnail(src, dst, thumbnail_size, encoding=None):
         save_kwargs = dict(encoding)
         if dst.suffix.lower() == ".png":
             save_kwargs = {"optimize": encoding["optimize"]}
-        _atomic_save(img, dst, **save_kwargs)
+        _atomic_save(img, dst, guard=guard, **save_kwargs)
+
+
+def _video_placeholder_image(thumbnail_size):
+    """The placeholder frame for a video no decoder could read: a play mark."""
+    from PIL import ImageDraw, ImageFont
+
+    img = Image.new("RGB", thumbnail_size, color=(50, 50, 50))
+    draw = ImageDraw.Draw(img)
+    center_x, center_y = thumbnail_size[0] // 2, thumbnail_size[1] // 2
+    triangle_size = min(thumbnail_size) // 4
+    points = [
+        (center_x - triangle_size // 2, center_y - triangle_size // 2),
+        (center_x - triangle_size // 2, center_y + triangle_size // 2),
+        (center_x + triangle_size // 2, center_y),
+    ]
+    draw.polygon(points, fill=(255, 255, 255))
+    try:
+        font = ImageFont.load_default()
+        text = "VIDEO"
+        bbox = draw.textbbox((0, 0), text, font=font)
+        text_width = bbox[2] - bbox[0]
+        draw.text(
+            (center_x - text_width // 2, center_y + triangle_size // 2 + 10),
+            text,
+            fill=(255, 255, 255),
+            font=font,
+        )
+    except (OSError, AttributeError):
+        pass
+    return img
+
+
+def _is_decode_error(exc):
+    """True when *exc* says the file itself cannot be decoded (#249 M4).
+
+    Pillow's decoder errors are OSErrors without an errno; a real I/O error
+    (disk full, a share gone) carries one and is worth another try.
+    """
+    if isinstance(exc, (UnidentifiedImageError, Image.DecompressionBombError)):
+        return True
+    if isinstance(exc, SyntaxError):
+        return True
+    return isinstance(exc, OSError) and exc.errno is None
 
 
 def _new_thumbnail_stats():
@@ -710,23 +774,51 @@ 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(output_path, rel_path, root_index, *, root_scope=None):
+    """``(thumbnail_url, ready)`` for *rel_path* under *output_path* (one
+    blocking stat). The Images listing, 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=`` and ``ready`` is
+    True. A missing one is still listed (without ``v=``): the serve route
+    makes it on first view and falls back when it cannot
+    (py/api/thumbnail_auto.py), so no card loads the full file because its
+    preview was not made yet. Media inside a nested ``thumbnails`` folder is
+    never previewed: ``(None, False)``, and its card shows the original.
+    """
+    rel_path = Path(rel_path)
+    if "thumbnails" in rel_path.parts[:-1]:
+        return None, False
+    thumb_rel = media_thumbnail_rel(rel_path)
     thumbnail_url = thumbnail_url_for(thumb_rel, Path(output_path) / thumb_rel)
+    ready = thumbnail_url is not None
     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
+    return thumbnail_url, ready
+
+
+def media_thumbnail_url(output_path, rel_path, root_index, *, root_scope=None):
+    """The thumbnail URL of ``media_thumbnail`` (None for nested previews)."""
+    return media_thumbnail(output_path, rel_path, root_index, root_scope=root_scope)[0]
 
 
 def _output_image_entry(
@@ -744,7 +836,7 @@ def _output_image_entry(
     extension = media_path.suffix.lower()
     is_video = extension in VIDEO_EXTENSIONS
 
-    thumbnail_url = media_thumbnail_url(
+    thumbnail_url, thumbnail_ready = media_thumbnail(
         output_path, rel_path, root_index, root_scope=root_scope
     )
     encoded_path = urllib.parse.quote(rel_path.as_posix(), safe="/")
@@ -763,6 +855,8 @@ def _output_image_entry(
         "root_index": root_index,
         "url": url,
         "thumbnail_url": thumbnail_url,
+        # Whether the preview file exists yet (the legacy gallery's dot, #249).
+        "thumbnail_ready": thumbnail_ready,
         "size": stat.st_size,
         "modified_time": stat.st_mtime,
         "extension": extension,
@@ -1682,6 +1776,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 +2008,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
                     ],
@@ -2003,6 +2104,8 @@ class ImageRoutesMixin:
         encoding=None,
         rebuild=False,
         stop=None,
+        guard=None,
+        track=None,
     ):
         """Create one thumbnail (blocking). Never raises; returns a result dict.
 
@@ -2010,7 +2113,11 @@ class ImageRoutesMixin:
         and touches nothing (a job's file boundary). ``rebuild`` writes a
         fresh preview again and never replaces an existing video preview with
         the placeholder. Every write is atomic, so a failure keeps the old
-        preview.
+        preview. ``guard()`` false just before the rename gives ``action:
+        "error"`` with ``held: True`` and writes nothing; ``track(process)``
+        is handed the ffmpeg process (both for automatic previews, #249). An
+        error result carries ``decode: True`` when the file itself cannot be
+        decoded (an I/O error does not).
         """
         result = {
             "file": src.name,
@@ -2027,22 +2134,43 @@ class ImageRoutesMixin:
             dst.parent.mkdir(parents=True, exist_ok=True)
             if is_video:
                 made = self._generate_video_thumbnail(
-                    src, dst, thumbnail_size, encoding, keep_existing=rebuild
+                    src,
+                    dst,
+                    thumbnail_size,
+                    encoding,
+                    keep_existing=rebuild,
+                    guard=guard,
+                    track=track,
                 )
                 if not made:
-                    return {**result, "action": "error", "error": "Video decode failed"}
+                    return {
+                        **result,
+                        "action": "error",
+                        "error": "Video decode failed",
+                        "decode": True,
+                    }
             else:
-                _write_image_thumbnail(src, dst, thumbnail_size, encoding)
+                _write_image_thumbnail(src, dst, thumbnail_size, encoding, guard)
             return {**result, "action": "generated"}
+        except PreviewWriteHeld:
+            return {**result, "action": "error", "error": PREVIEW_HELD, "held": True}
         except UnidentifiedImageError:
-            return {**result, "action": "error", "error": "Not a valid image file"}
+            return {
+                **result,
+                "action": "error",
+                "error": "Not a valid image file",
+                "decode": True,
+            }
         except PermissionError as e:
             message = _preview_write_error(e)
             self.logger.warning(f"Thumbnail failed for {src.name}: {message}")
             return {**result, "action": "error", "error": message}
         except Exception as e:
             self.logger.warning(f"Thumbnail failed for {src.name}", exc_info=True)
-            return {**result, "action": "error", "error": _safe_error(e)}
+            failed = {**result, "action": "error", "error": _safe_error(e)}
+            if _is_decode_error(e):
+                failed["decode"] = True
+            return failed
 
     async def _emit_sse(self, response, event, data) -> bool:
         """Write one SSE frame. Returns False once the client has gone away."""
@@ -2061,147 +2189,59 @@ class ImageRoutesMixin:
         thumbnail_size,
         encoding=None,
         keep_existing=False,
+        *,
+        guard=None,
+        track=None,
     ):
         """Generate thumbnail from video file. Returns True if successful.
 
-        ``encoding`` is one of THUMBNAIL_ENCODINGS (medium when None); it
-        encodes the OpenCV frame, ffmpeg's frame (as ``-q:v``) and the
-        placeholder alike. Both decoders fit the frame inside
-        ``thumbnail_size`` keeping its aspect ratio. With ``keep_existing`` an
-        existing preview is kept, and the run fails, when no frame could be
-        decoded, rather than replaced by the placeholder. A PermissionError
-        while replacing the preview propagates (``_generate_one`` names it).
+        Tries OpenCV, then ffmpeg, then writes the placeholder frame: a
+        decoder that cannot open or read the clip (an AV1 file OpenCV was
+        built without) falls through to the next one (#249 I1). ``encoding``
+        is one of THUMBNAIL_ENCODINGS (medium when None); it encodes the
+        OpenCV frame, ffmpeg's frame (as ``-q:v``) and the placeholder alike.
+        Both decoders fit the frame inside ``thumbnail_size`` keeping its
+        aspect ratio. With ``keep_existing`` an existing preview is kept, and
+        the run fails, when no frame could be decoded, rather than replaced
+        by the placeholder. A PermissionError while replacing the preview
+        propagates (``_generate_one`` names it), as does PreviewWriteHeld
+        when ``guard()`` refuses the rename. ``track(process)`` is handed the
+        ffmpeg process so a shutdown can kill it.
         """
         encoding = encoding or DEFAULT_THUMBNAIL_ENCODING
         try:
-            # Try using OpenCV first (most reliable)
-            try:
-                import cv2
-
-                cap = cv2.VideoCapture(str(video_path))
-                if not cap.isOpened():
-                    return False
-
-                frame_count = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
-                target_frame = max(1, int(frame_count * 0.1))
-                cap.set(cv2.CAP_PROP_POS_FRAMES, target_frame)
-
-                ret, frame = cap.read()
-                cap.release()
-
-                if not ret or frame is None:
-                    return False
-
-                frame_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
-
-                img = Image.fromarray(frame_rgb)
-                img.thumbnail(thumbnail_size, Image.Resampling.LANCZOS)
-                _atomic_save(img, thumbnail_path, "JPEG", **encoding)
+            frame = self._opencv_frame(video_path, thumbnail_size)
+            if frame is not None:
+                _atomic_save(frame, thumbnail_path, "JPEG", guard=guard, **encoding)
                 self.logger.debug(
                     f"Generated video thumbnail using OpenCV: {thumbnail_path}"
                 )
                 return True
-
-            except ImportError:
-                pass
-
-            # Fallback to ffmpeg
-            try:
-                import subprocess
-
-                tmp_path = _temp_sibling(thumbnail_path)
-                cmd = [
-                    "ffmpeg",
-                    "-i",
-                    str(video_path),
-                    "-ss",
-                    "00:00:01",
-                    "-vframes",
-                    "1",
-                    "-vf",
-                    f"scale={thumbnail_size[0]}:{thumbnail_size[1]}"
-                    ":force_original_aspect_ratio=decrease",
-                    "-q:v",
-                    _ffmpeg_qscale(encoding),
-                    "-y",
-                    str(tmp_path),
-                ]
-
-                result = subprocess.run(cmd, capture_output=True, text=True, timeout=30)
-                if result.returncode == 0 and tmp_path.exists():
-                    os.replace(tmp_path, thumbnail_path)
-                    self.logger.debug(
-                        f"Generated video thumbnail using ffmpeg: {thumbnail_path}"
-                    )
-                    return True
-                else:
-                    self.logger.warning(
-                        f"ffmpeg failed for {video_path}: {result.stderr}"
-                    )
-
-            except (ImportError, subprocess.TimeoutExpired, FileNotFoundError):
-                pass
-            finally:
-                try:
-                    tmp_path.unlink()
-                except (NameError, OSError):
-                    pass
-
+            if self._ffmpeg_frame(
+                video_path, thumbnail_path, thumbnail_size, encoding, guard, track
+            ):
+                return True
             if keep_existing and thumbnail_path.exists():
                 self.logger.warning(
                     f"No frame decoded for {video_path.name}; keeping its preview"
                 )
                 return False
-
-            # Last resort: create a placeholder thumbnail
+            # Last resort: the placeholder frame.
             try:
-                from PIL import ImageDraw, ImageFont
-
-                img = Image.new("RGB", thumbnail_size, color=(50, 50, 50))
-                draw = ImageDraw.Draw(img)
-
-                center_x, center_y = thumbnail_size[0] // 2, thumbnail_size[1] // 2
-                triangle_size = min(thumbnail_size) // 4
-
-                points = [
-                    (center_x - triangle_size // 2, center_y - triangle_size // 2),
-                    (center_x - triangle_size // 2, center_y + triangle_size // 2),
-                    (center_x + triangle_size // 2, center_y),
-                ]
-                draw.polygon(points, fill=(255, 255, 255))
-
-                try:
-                    font = ImageFont.load_default()
-                    text = "VIDEO"
-                    bbox = draw.textbbox((0, 0), text, font=font)
-                    text_width = bbox[2] - bbox[0]
-                    draw.text(
-                        (
-                            center_x - text_width // 2,
-                            center_y + triangle_size // 2 + 10,
-                        ),
-                        text,
-                        fill=(255, 255, 255),
-                        font=font,
-                    )
-                except (OSError, AttributeError):
-                    pass
-
-                _atomic_save(img, thumbnail_path, "JPEG", **encoding)
+                img = _video_placeholder_image(thumbnail_size)
+                _atomic_save(img, thumbnail_path, "JPEG", guard=guard, **encoding)
                 self.logger.debug(
                     f"Generated placeholder video thumbnail: {thumbnail_path}"
                 )
                 return True
-
-            except PermissionError:
+            except (PermissionError, PreviewWriteHeld):
                 raise
             except Exception as e:
                 self.logger.warning(
                     f"Failed to create placeholder thumbnail for {video_path}: {e}"
                 )
                 return False
-
-        except PermissionError:
+        except (PermissionError, PreviewWriteHeld):
             raise
         except Exception as e:
             self.logger.error(
@@ -2209,6 +2249,88 @@ class ImageRoutesMixin:
             )
             return False
 
+    def _opencv_frame(self, video_path, thumbnail_size):
+        """A frame about 10% in, fitted to *thumbnail_size*, or None (no
+        OpenCV, or it could not open or read the clip)."""
+        try:
+            import cv2
+        except ImportError:
+            return None
+        try:
+            cap = cv2.VideoCapture(str(video_path))
+            try:
+                if not cap.isOpened():
+                    return None
+                frame_count = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
+                cap.set(cv2.CAP_PROP_POS_FRAMES, max(1, int(frame_count * 0.1)))
+                ret, frame = cap.read()
+            finally:
+                cap.release()
+            if not ret or frame is None:
+                return None
+            img = Image.fromarray(cv2.cvtColor(frame, cv2.COLOR_BGR2RGB))
+            img.thumbnail(thumbnail_size, Image.Resampling.LANCZOS)
+            return img
+        except Exception as e:
+            self.logger.debug(f"OpenCV could not read {video_path}: {e}")
+            return None
+
+    def _ffmpeg_frame(
+        self, video_path, thumbnail_path, thumbnail_size, encoding, guard, track
+    ):
+        """Write ffmpeg's frame at 1 s into *thumbnail_path*; True on success."""
+        import subprocess  # nosec B404 - fixed argv, no shell
+
+        tmp_path = _temp_sibling(thumbnail_path)
+        cmd = [
+            "ffmpeg",
+            "-i",
+            str(video_path),
+            "-ss",
+            "00:00:01",
+            "-vframes",
+            "1",
+            "-vf",
+            f"scale={thumbnail_size[0]}:{thumbnail_size[1]}"
+            ":force_original_aspect_ratio=decrease",
+            "-q:v",
+            _ffmpeg_qscale(encoding),
+            "-y",
+            str(tmp_path),
+        ]
+        try:
+            process = subprocess.Popen(  # nosec B603 B607 - fixed argv
+                cmd,
+                stdout=subprocess.PIPE,
+                stderr=subprocess.PIPE,
+                text=True,
+            )
+            if track is not None:
+                track(process)
+            try:
+                _out, stderr = process.communicate(timeout=30)
+            except subprocess.TimeoutExpired:
+                process.kill()
+                process.communicate()
+                return False
+            if process.returncode == 0 and tmp_path.exists():
+                _replace_unless_held(tmp_path, thumbnail_path, guard)
+                self.logger.debug(
+                    f"Generated video thumbnail using ffmpeg: {thumbnail_path}"
+                )
+                return True
+            self.logger.warning(f"ffmpeg failed for {video_path}: {stderr}")
+            return False
+        except (FileNotFoundError, OSError) as e:
+            if isinstance(e, PermissionError) and tmp_path.exists():
+                raise
+            return False
+        finally:
+            try:
+                tmp_path.unlink()
+            except OSError:
+                pass
+
     async def clear_thumbnails(self, request: web.Request) -> web.Response:
         """Clear generated thumbnails across gallery roots, preserving originals.
 
@@ -2218,7 +2340,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,843 @@
+"""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``), 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 is admitted on the event loop
+  (``_PreviewGate``): at most ``ON_DEMAND_LIMIT`` previews are written at once,
+  on this API's own small pool (never the default executor), one flight runs
+  per file, and the requests waiting are capped (``MAX_WAITING``); past the
+  cap a request gets its fallback at once. Inside the admission the preview is
+  mapped back to its media file under the same root (the request text is
+  checked first: no traversal, drive, UNC or empty segment; then a regular
+  file inside the root, never through a link), made and served. When it
+  cannot be made an image's original is served and a video's placeholder
+  frame, both ``Cache-Control: no-store``: a video is never sent to an
+  ``<img>``. A file that cannot be decoded is remembered for this session
+  (keyed by its path, size and mtime) and counted for the System Overview
+  (``failed`` in ``/images/thumbnails/status``); I/O errors and a preview
+  another program holds open (Windows) are not remembered. The recorded
+  history's thumbnail route goes through the same admission and flight key.
+- **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``). The holds are checked when a
+write starts (after its slot) and again just before its rename.
+
+Shutdown: ``close_auto_thumbnails`` stops the worker at a file boundary,
+kills a running ffmpeg, waits a bounded time (and logs when the wait ran
+out) and shuts the view pool down; after the app's cleanup
+(``final=True``) nothing starts a worker again. Importing this module again
+(a package reload) stops the workers the previous import left.
+"""
+
+import asyncio
+import collections
+import functools
+import io
+import os
+import threading
+from concurrent.futures import ThreadPoolExecutor
+from pathlib import Path
+
+from aiohttp import web
+
+from .images import (
+    GALLERY_IMAGE_EXTENSIONS,
+    MEDIA_EXTENSIONS,
+    THUMBNAIL_CLEAR_TITLE,
+    THUMBNAIL_ENCODINGS,
+    VIDEO_EXTENSIONS,
+    _has_traversal,
+    _is_within,
+    _thumbnail_is_fresh,
+    _video_placeholder_image,
+    _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"]
+# Previews views may write at once (the size of the view pool).
+ON_DEMAND_LIMIT = 3
+# View requests that may wait on the gate; one more gets its fallback at once.
+MAX_WAITING = 64
+# How long a pool thread waits for the capture worker's flight of the same file.
+FLIGHT_WAIT_SECONDS = 20
+# How long close() waits for the worker's batch in hand (ffmpeg's own limit).
+CLOSE_WAIT_SECONDS = 30
+# 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}
+)
+WORKER_NAME = "PromptManagerThumbnailWorker"
+POOL_PREFIX = "pm-preview"
+# Identifies this import's workers: a reload stops the ones it did not start.
+_MODULE_TOKEN = object()
+
+READY, FAILED, BUSY, DEFERRED = "ready", "failed", "busy", "deferred"
+# What the gate answers a request past MAX_WAITING.
+OVER = object()
+
+
+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 flight_key(root, rel_parts):
+    """The single-flight key of ``root/rel_parts``: its folder resolved, case
+    folded. The serve route, the history route and the worker share it."""
+    path = Path(root).joinpath(*rel_parts)
+    return os.path.normcase(os.path.join(os.path.realpath(path.parent), path.name))
+
+
+def _dst_key(dst):
+    return flight_key(Path(dst).parent, (Path(dst).name,))
+
+
+@functools.lru_cache(maxsize=1)
+def video_placeholder_jpeg():
+    """The placeholder frame as JPEG bytes (a video whose preview is missing)."""
+    buffer = io.BytesIO()
+    _video_placeholder_image(AUTO_THUMBNAIL_SIZE).save(
+        buffer, "JPEG", **AUTO_THUMBNAIL_ENCODING
+    )
+    return buffer.getvalue()
+
+
+def _is_preview_request(filepath):
+    """Text check only: ``thumbnails/<...>/<name>_thumb<lower-case ext>``.
+
+    Refuses, as text and before any path is joined, what a joined path could
+    take as an anchor: a backslash, a drive or stream colon, and empty, ``.``
+    or ``..`` segments (so a leading ``/`` and ``//server/share`` too).
+    """
+    if "\\" in filepath or ":" in filepath:
+        return False
+    parts = filepath.split("/")
+    if len(parts) < 2 or parts[0] != "thumbnails":
+        return False
+    if any(part in ("", ".", "..") for part in parts) 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 _candidates(parts):
+    """``(folder parts, base, extensions)`` of a checked preview request."""
+    stem, extension = os.path.splitext(parts[-1])
+    wanted = [extension] + (sorted(VIDEO_EXTENSIONS) if extension == ".jpg" else [])
+    return parts[1:-1], stem[: -len("_thumb")], wanted
+
+
+def _quick_source(root, parts):
+    """``(src, is_video)`` by exact name only (no folder listing), or None."""
+    root = Path(root)
+    folder_parts, base, wanted = _candidates(parts)
+    folder = root.joinpath(*folder_parts)
+    for extension in wanted:
+        for name in (base + extension, base + extension.upper()):
+            path = folder / name
+            if _regular_file(path) and _is_within(path, root):
+                return path, extension in VIDEO_EXTENSIONS
+    return None
+
+
+def _thumbnail_source(root, parts):
+    """``(src, dst, is_video)`` for the preview *parts* under *root*, or None.
+
+    *parts* passed ``_is_preview_request``. The source has the preview's name
+    without ``_thumb`` and an extension that previews as the requested one
+    (any case: exact names first, then one listing of the folder); a ``.jpg``
+    may also be a video's. It must be a regular file inside *root*. ``dst``
+    is None when the preview's folder resolves outside ``root/thumbnails`` (a
+    link): nothing is written there.
+    """
+    root = Path(root)
+    folder_parts, base, wanted = _candidates(parts)
+    quick = _quick_source(root, parts)
+    if quick is not None:
+        found = quick[0]
+    else:
+        found = _scan_folder(root.joinpath(*folder_parts), base, set(wanted))
+        if found is None or not _is_within(found, root):
+            return None
+    rel = Path(*folder_parts, 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
+    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
+
+
+def _stop_stale_workers():
+    """Stop capture workers a previous import of this module started."""
+    for thread in threading.enumerate():
+        token = getattr(thread, "pm_preview_token", None)
+        close = getattr(thread, "pm_preview_close", None)
+        if token is not None and token is not _MODULE_TOKEN and close is not None:
+            close()
+
+
+class AutoThumbnailer:
+    """Single-flight preview writes and the capture worker (thread-safe).
+
+    ``generate(src, dst, is_video, *, stop, guard, track)`` writes one
+    preview and returns ``_generate_one``'s result dict; ``may_write()``
+    says whether automatic previews may be written now. Views call
+    ``ensure`` on the view pool, the monitor calls ``enqueue`` from its
+    thread.
+    """
+
+    def __init__(
+        self,
+        generate,
+        may_write,
+        logger,
+        *,
+        workers=_worker_threads,
+        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._failures = collections.OrderedDict()
+        self._queue = collections.deque()
+        self._queued = set()
+        self._processes = []
+        self._worker = None
+        # True from a worker's start until it leaves (under the lock), so an
+        # idle worker can exit and the next enqueue starts a fresh one.
+        self._worker_running = False
+        self._closed = False
+        self._busy = False
+        self._batch = 0
+
+    @property
+    def failure_count(self):
+        with self._changed:
+            return len(self._failures)
+
+    @property
+    def closed(self):
+        return self._closed
+
+    def _allowed(self):
+        return not self._closed and self._may_write()
+
+    def _track(self, process):
+        """Keep an ffmpeg process so close() can kill it."""
+        with self._changed:
+            self._processes = [p for p in self._processes if p.poll() is None]
+            self._processes.append(process)
+            closed = self._closed
+        if closed:
+            process.kill()
+
+    # ── one preview ───────────────────────────────────────────────────
+
+    def ensure(self, src, dst, is_video):
+        """Make *dst* for *src* unless it is fresh; returns the outcome.
+
+        ``ready`` (the preview is there), ``failed`` (the file cannot be
+        decoded, remembered), ``busy`` (an I/O error, the file is gone, or
+        another flight did not finish) or ``deferred`` (a hold refuses
+        automatic writes, checked when the write starts and before its
+        rename). 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 = _dst_key(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()
+        if flight is not None:
+            flight.wait(FLIGHT_WAIT_SECONDS)
+            if _thumbnail_is_fresh(src, dst):
+                return READY
+            with self._changed:
+                return FAILED if fingerprint in self._failures else BUSY
+        try:
+            return self._lead(src, dst, is_video, fingerprint)
+        finally:
+            with self._changed:
+                self._inflight.pop(key, None)
+                self._changed.notify_all()
+            done.set()
+
+    def _lead(self, src, dst, is_video, fingerprint):
+        if not self._allowed():
+            return DEFERRED
+        if _thumbnail_is_fresh(src, dst):
+            return READY
+        result = self._generate(
+            src,
+            dst,
+            is_video,
+            stop=lambda: self._closed,
+            guard=self._allowed,
+            track=self._track,
+        )
+        if result.get("action") in ("generated", "skipped"):
+            return READY
+        if result.get("held"):
+            return DEFERRED
+        if not result.get("decode"):
+            return BUSY  # an I/O error says nothing about the file
+        error = result.get("error") or "failed"
+        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
+
+    # ── 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 = _dst_key(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 not self._worker_running:
+                self._start_worker_locked()
+            self._changed.notify_all()
+        return True
+
+    def _start_worker_locked(self):
+        """Start the worker (lock held). A start that fails leaves the file
+        queued and the next enqueue tries again."""
+        worker = threading.Thread(target=self._run, name=WORKER_NAME, daemon=True)
+        worker.pm_preview_token = _MODULE_TOKEN
+        worker.pm_preview_close = self.close
+        self._worker_running = True
+        try:
+            worker.start()
+        except RuntimeError:
+            self._worker_running = False
+            self._busy = False
+            self._logger.warning(
+                "The automatic preview worker could not start; the next output "
+                "tries again",
+                exc_info=True,
+            )
+            return
+        self._worker = worker
+
+    def _run(self):
+        try:
+            self._drain()
+        except Exception:
+            self._logger.warning("The automatic preview worker stopped", exc_info=True)
+        finally:
+            # Whatever ended the loop, the next enqueue starts a fresh worker.
+            with self._changed:
+                self._worker_running = False
+                self._busy = False
+                self._batch = 0
+                self._changed.notify_all()
+
+    def _drain(self):
+        while True:
+            with self._changed:
+                if self._closed or not self._queue:
+                    # Idle or closed: leave, so no thread outlives its work.
+                    self._busy = False
+                    self._worker_running = 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)
+        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 at its next file boundary (queued files are
+        dropped), kill a running ffmpeg and wait up to CLOSE_WAIT_SECONDS;
+        a worker still running after that is logged."""
+        with self._changed:
+            self._closed = True
+            self._queue.clear()
+            self._queued.clear()
+            processes, self._processes = self._processes, []
+            self._changed.notify_all()
+            worker = self._worker
+        for process in processes:
+            try:
+                process.kill()
+            except OSError:
+                pass
+        if worker is None or worker is threading.current_thread():
+            return
+        worker.join(CLOSE_WAIT_SECONDS)
+        if worker.is_alive():
+            self._logger.warning(
+                "The automatic preview worker is still running after "
+                f"{CLOSE_WAIT_SECONDS} s; it stops at its next file"
+            )
+
+    # ── 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_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 _PreviewGate:
+    """Admission of view requests on the event loop (#249 I2).
+
+    Waiters are coroutines: at most ``limit`` previews are written at once
+    on the view pool, one flight runs per key, and past ``MAX_WAITING``
+    waiting requests ``run`` answers ``OVER`` at once.
+    """
+
+    def __init__(self, limit, logger):
+        self.loop = asyncio.get_running_loop()
+        self._slots = asyncio.Semaphore(limit)
+        self._flights = {}
+        self._tasks = set()
+        self._logger = logger
+        self._tick = asyncio.Event()
+        self.waiting = 0
+
+    def _changed(self):
+        tick, self._tick = self._tick, asyncio.Event()
+        tick.set()
+
+    async def settled(self, predicate):
+        """Return once ``predicate()`` holds (re-checked at each change)."""
+        while not predicate():
+            await self._tick.wait()
+
+    async def run(self, key, work, pool):
+        """``work()``'s answer for *key* (shared by every request of the
+        flight), None when it raised, or OVER past the cap."""
+        if self.waiting >= MAX_WAITING:
+            return OVER
+        flight = self._flights.get(key)
+        if flight is None:
+            flight = self.loop.create_future()
+            self._flights[key] = flight
+            task = self.loop.create_task(self._lead(key, flight, work, pool))
+            self._tasks.add(task)
+            task.add_done_callback(self._tasks.discard)
+        self.waiting += 1
+        self._changed()
+        try:
+            return await asyncio.shield(flight)
+        finally:
+            self.waiting -= 1
+            self._changed()
+
+    async def _lead(self, key, flight, work, pool):
+        result = None
+        try:
+            async with self._slots:
+                result = await self.loop.run_in_executor(pool, work)
+        except Exception:
+            self._logger.warning("Automatic preview failed", exc_info=True)
+        finally:
+            self._flights.pop(key, None)
+            if not flight.done():
+                flight.set_result(result)
+
+
+class AutoThumbnailMixin:
+    """Automatic previews on the API: views, the monitor's queue, scans."""
+
+    _auto_thumbnailer = None
+    _auto_thumbnails_shut = False
+    _preview_gate_obj = None
+    _preview_pool_obj = None
+    _auto_thumbnailer_lock = threading.Lock()
+
+    def _auto_thumbnails(self):
+        """This API's AutoThumbnailer, created on first use (no worker after
+        the app's cleanup)."""
+        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,
+                    start_worker=not self._auto_thumbnails_shut,
+                )
+            return self._auto_thumbnailer
+
+    def close_auto_thumbnails(self, final=False):
+        """Stop the capture worker and the view pool, and forget them.
+
+        Bounded (see AutoThumbnailer.close). ``final`` (the app's cleanup)
+        also keeps a late monitor event from starting a worker again; a
+        test reset passes False and the next use starts fresh ones.
+        """
+        with AutoThumbnailMixin._auto_thumbnailer_lock:
+            if final:
+                self._auto_thumbnails_shut = True
+            thumbnailer, self._auto_thumbnailer = self._auto_thumbnailer, None
+            pool, self._preview_pool_obj = self._preview_pool_obj, None
+            self._preview_gate_obj = None
+        if thumbnailer is not None:
+            thumbnailer.close()
+        if pool is not None:
+            pool.shutdown(wait=False, cancel_futures=True)
+
+    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, *, stop, guard, track):
+        return self._generate_one(
+            src,
+            dst,
+            AUTO_THUMBNAIL_SIZE,
+            is_video,
+            encoding=AUTO_THUMBNAIL_ENCODING,
+            stop=stop,
+            guard=guard,
+            track=track,
+        )
+
+    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
+
+    # ── the view admission ────────────────────────────────────────────
+
+    def _preview_gate(self):
+        """The gate of the running loop (a new loop gets a new one)."""
+        gate = self._preview_gate_obj
+        if gate is None or gate.loop is not asyncio.get_running_loop():
+            gate = self._preview_gate_obj = _PreviewGate(ON_DEMAND_LIMIT, self.logger)
+        return gate
+
+    def _preview_pool(self):
+        with AutoThumbnailMixin._auto_thumbnailer_lock:
+            if self._preview_pool_obj is None:
+                self._preview_pool_obj = ThreadPoolExecutor(
+                    ON_DEMAND_LIMIT, thread_name_prefix=POOL_PREFIX
+                )
+            return self._preview_pool_obj
+
+    def _preview_pool_threads(self):
+        pool = self._preview_pool_obj
+        return [] if pool is None else list(getattr(pool, "_threads", ()))
+
+    def _video_placeholder_bytes(self):
+        return video_placeholder_jpeg()
+
+    async def _admit_preview(self, src, dst, is_video):
+        """Make one preview through the view admission (the history route).
+
+        Shares the serve route's flight key; every flight answers the
+        ``ensure`` outcome (or None, or OVER), whichever route started it.
+        """
+        return await self._preview_gate().run(
+            flight_key(Path(dst).parent, (Path(dst).name,)),
+            functools.partial(self._ensure_view, src, dst, is_video),
+            self._preview_pool(),
+        )
+
+    def _ensure_view(self, src, dst, is_video):
+        """``ensure``, reporting a preview removed under us as busy."""
+        outcome = self._auto_thumbnails().ensure(src, dst, is_video)
+        if outcome == READY and not Path(dst).is_file():
+            return BUSY  # removed under us: the fallback this once
+        return outcome
+
+    # ── on first view ─────────────────────────────────────────────────
+
+    def _preview_on_demand(self, root, parts):
+        """The ``ensure`` outcome for the preview *parts* under *root*
+        (``deferred`` when its folder is linked out of the root), or None
+        when no media file previews as *parts*."""
+        located = _thumbnail_source(root, parts)
+        if located is None:
+            return None
+        src, dst, is_video = located
+        if dst is None:
+            return DEFERRED
+        return self._ensure_view(src, dst, is_video)
+
+    def _preview_response(self, outcome, src, dst, is_video):
+        if outcome == READY and dst is not None:
+            return self._file_response(dst)
+        if is_video:
+            return web.Response(
+                body=video_placeholder_jpeg(),
+                content_type="image/jpeg",
+                headers={"Cache-Control": "no-store"},
+            )
+        response = web.FileResponse(src)
+        response.headers["Cache-Control"] = "no-store"
+        return response
+
+    async def _serve_missing_thumbnail(self, output_dirs, filepath):
+        """The made preview, or its fallback; None when *filepath* names no
+        preview of a media file in *output_dirs*.
+
+        The flight answers only the outcome (a history flight may be the one
+        joined); the file it previews is derived here: the preview path is
+        the request's own, and a fallback looks its source up again on the
+        view pool (bounded by it, never the default executor).
+        """
+        if _has_traversal(filepath) or not _is_preview_request(filepath):
+            return None
+        parts = tuple(filepath.split("/"))
+        gate = self._preview_gate()
+        pool = self._preview_pool()
+        for root in output_dirs:
+            root = Path(root)
+            outcome = await gate.run(
+                flight_key(root, parts),
+                functools.partial(self._preview_on_demand, root, parts),
+                pool,
+            )
+            if outcome is None:
+                continue
+            if outcome is OVER:
+                # Past the cap: exact names only, no listing, no write.
+                quick = _quick_source(root, parts)
+                if quick is None:
+                    continue
+                return self._preview_response(BUSY, quick[0], None, quick[1])
+            dst = root.joinpath(*parts)
+            if outcome == READY:
+                return self._file_response(dst)
+            located = await gate.loop.run_in_executor(
+                pool, _thumbnail_source, root, parts
+            )
+            if located is None:
+                continue
+            src, _dst, is_video = located
+            return self._preview_response(outcome, src, None, is_video)
+        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,
+        already queued, or after the app's cleanup). Never writes on the
+        caller's thread.
+        """
+        if self._auto_thumbnails_shut:
+            return "skipped"
+        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
+            dst = root / media_thumbnail_rel(rel)
+            if not _is_within(root / "thumbnails", root) or not _is_within(
+                dst, root / "thumbnails"
+            ):
+                return None
+            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
+
+
+# A package reload re-runs this module: stop the workers the old one left.
+_stop_stale_workers()

tests/browser/server.py

@@ -386,6 +386,8 @@ def create_app(root: Path) -> web.Application:
     async def reset(request: web.Request) -> web.Response:
         nonlocal history_fixture
         request_holds.release()
+        # No preview queued by the last test may land in the new fixture.
+        await api._run_in_executor(api.close_auto_thumbnails)
         library_editor_commands.reset()
         data = await request.json() if request.can_read_body else {}
         if not isinstance(data, dict):

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,45 @@ 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();
+    // #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.__oldInspectorCover.isConnected)).toBe(true);
+    expect(await page.evaluate(() => window.__oldInspectorCover.getAttribute('src'))).toBeNull();
+});
+
+/** A listing that names no preview for its videos (the fallback a card and the cover keep for one). */
+async function videosWithoutPreview(page) {
+    await page.route(url => url.pathname === '/prompt_manager/images/output', async route => {
+        const response = await route.fetch();
+        const body = await response.json();
+        if (Array.isArray(body.images)) body.images = body.images.map(image => image.is_video ? { ...image, thumbnail_url: null } : image);
+        await route.fulfill({ response, json: body });
+    });
+}
+
+test('a video listed without a preview keeps its own element, released on refresh', async ({ page, request }) => {
+    await expect(await request.post('/__test__/reset', { data: { include_video: true } })).toBeOK();
+    await videosWithoutPreview(page);
+    await page.goto(workspace);
+    const card = cards(page).filter({ hasText: 'clip.webm' });
+    await expect(card.locator('.media-art > .image-open > video')).toHaveCount(1);
+    await expect(card.locator('img')).toHaveCount(0);
+    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; });
+    await card.locator('video').evaluate(element => { window.__oldCardVideo = 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.__oldInspectorVideo.isConnected && !window.__oldCardVideo.isConnected)).toBe(true);
+    expect(await page.evaluate(() => [window.__oldInspectorVideo.getAttribute('src'), window.__oldCardVideo.getAttribute('src')])).toEqual([null, null]);
 });
 
 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,135 @@
+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([]);
+});
+
+test('a video whose preview cannot be loaded swaps back to its own element', async ({ page, request }) => {
+    await expect(await request.post('/__test__/reset', { data: { include_video: true } })).toBeOK();
+    await page.route(url => url.pathname.endsWith('/thumbnails/clip_thumb.jpg'), route => route.fulfill({ status: 404, body: '' }));
+    await page.goto(images);
+    const card = cards(page).filter({ hasText: 'clip.webm' });
+    await expect(card.locator('.media-art > .image-open > video')).toHaveCount(1);
+    await expect(card.locator('.media-art > .image-open > img')).toHaveCount(0);
+    await card.getByRole('button', { name: 'Inspect clip.webm', exact: true }).click();
+    await expect(details(page).locator('video.detail-cover')).toHaveCount(1);
+    await expect(details(page).locator('img.detail-cover')).toHaveCount(0);
+});
+
+test('an image listed without a preview (a nested thumbnails folder) shows its original', async ({ page, request }) => {
+    await expect(await request.post('/__test__/reset')).toBeOK();
+    // M10: such files are listed with a null thumbnail_url; the card falls back to the original as before.
+    await page.route(url => url.pathname === '/prompt_manager/images/output', async route => {
+        const response = await route.fetch();
+        const body = await response.json();
+        if (Array.isArray(body.images)) body.images = body.images.map(image => ({ ...image, thumbnail_url: null }));
+        await route.fulfill({ response, json: body });
+    });
+    const seen = recordMedia(page);
+    await page.goto(images);
+    await expect(cards(page)).toHaveCount(2);
+    await settled(page);
+    expect(seen.length).toBeGreaterThan(0);
+    expect(seen.every(item => !item.preview)).toBe(true);
+});

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,30 @@ 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 });
+});
+
+test("names the server quotes (parentheses, '!*') still match their original and preview (#249 M11)", () => {
+    const toRecord = contract('toInspectionRecord');
+    const entry = { root_index: 0, path: "folder/a (1)!'*.png", filename: "a (1)!'*.png", root_scope: scopeA,
+        url: `/prompt_manager/images/serve/folder/a%20%281%29%21%27%2A.png?root=0&root_scope=${scopeA}`,
+        thumbnail_url: `/prompt_manager/images/serve/thumbnails/folder/a%20%281%29%21%27%2A_thumb.png?root=0&root_scope=${scopeA}` };
+    const record = toRecord(envelope(entry), entry, extractGenerationFacts);
+    assert.equal(record.download.url, entry.url);
+    assert.deepEqual(record.preview, { url: entry.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_admin.py

@@ -79,6 +79,8 @@ class AdminAPITestCase(AioHTTPTestCase):
 
     async def tearDownAsync(self):
         await super().tearDownAsync()  # closes the aiohttp test client
+        # The automatic preview worker writes into the temp tree: stop it first.
+        self.api.close_auto_thumbnails()
         GalleryConfig.MONITORING_DIRECTORIES = self._orig_gallery_dirs
         PromptManagerConfig.update_config(self._orig_pm_config)
         for key, value in self._orig_env.items():

tests/test_api_autotag.py

@@ -131,6 +131,8 @@ class AutotagAPITestCase(AioHTTPTestCase):
 
     async def tearDownAsync(self):
         await super().tearDownAsync()  # closes the aiohttp test client
+        # The automatic preview worker writes into the temp tree: stop it first.
+        self.api.close_auto_thumbnails()
         self._service_patch.stop()
         if model_gate is not None:
             leaked = model_gate.holder()

tests/test_api_backup_restore.py

@@ -62,6 +62,8 @@ class BackupRestoreTestCase(AioHTTPTestCase):
 
     async def tearDownAsync(self):
         await super().tearDownAsync()  # closes the aiohttp test client
+        # The automatic preview worker writes into the temp tree: stop it first.
+        self.api.close_auto_thumbnails()
         PromptModel.close_all_instances()  # restore may have swapped api.db
         assert_closed(self, self.tmpdir)
         for name in os.listdir(self.tmpdir):

tests/test_api_core.py

@@ -384,6 +384,8 @@ class CoreRoutesTestCase(AioHTTPTestCase):
 
     async def tearDownAsync(self):
         await super().tearDownAsync()  # closes the aiohttp test client
+        # The automatic preview worker writes into the temp tree: stop it first.
+        self.api.close_auto_thumbnails()
         self.api.db.close_all()
         assert_closed(self, self.tmpdir)
         shutil.rmtree(self.tmpdir, ignore_errors=True)

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())
 
@@ -1111,6 +1116,73 @@ class HistoryOutputAPITests(ImageAPITestCase):
         response = await self.current(descriptor["thumbnail_url"])
         self.assert_error(response, await response.json(), 403, "not-authorized")
 
+    async def test_missing_preview_is_never_made_through_a_linked_out_folder(self):
+        original = make_png(self.output_dir / "image.png", size=(64, 64))
+        elsewhere = self.temp_dir / "elsewhere"
+        elsewhere.mkdir()
+        (self.output_dir / "thumbnails").symlink_to(elsewhere, target_is_directory=True)
+        start = self.capture([self.ref()])
+        descriptor = await self.media(start)
+        response = await self.current(descriptor["thumbnail_url"])
+        self.assertEqual(response.status, 200)
+        self.assertEqual(await response.read(), original.read_bytes())
+        self.assertEqual(os.listdir(elsewhere), [])
+
+    async def test_missing_preview_is_never_made_through_a_linked_sub_folder(self):
+        # M6: every link segment of the preview's folder is checked.
+        original = make_png(self.output_dir / "sub" / "image.png", size=(64, 64))
+        elsewhere = self.temp_dir / "elsewhere"
+        elsewhere.mkdir()
+        (self.output_dir / "thumbnails").mkdir()
+        (self.output_dir / "thumbnails" / "sub").symlink_to(
+            elsewhere, target_is_directory=True
+        )
+        start = self.capture([self.ref(subfolder="sub")])
+        descriptor = await self.media(start)
+        response = await self.current(descriptor["thumbnail_url"])
+        self.assertEqual(response.status, 200)
+        self.assertEqual(await response.read(), original.read_bytes())
+        self.assertEqual(os.listdir(elsewhere), [])
+
+    async def test_a_video_preview_that_cannot_be_made_is_the_placeholder(self):
+        # I1: never the video bytes inside an <img>.
+        from py.api.jobs import RESTORE_JOB_KIND
+
+        (self.output_dir / "clip.webm").write_bytes(b"AV1 stream no decoder here")
+        start = self.capture([self.ref(filename="clip.webm")])
+        descriptor = await self.media(start)
+        gate = asyncio.Event()
+
+        async def runner(job):
+            await gate.wait()
+
+        job = self.api._jobs.start(
+            RESTORE_JOB_KIND,
+            "restore",
+            admission="records",
+            cancellable=False,
+            runner=runner,
+        )
+        try:
+            response = await self.current(descriptor["thumbnail_url"])
+            self.assertEqual(response.status, 200)
+            self.assertEqual(response.headers["Content-Type"], "image/jpeg")
+            self.assertTrue((await response.read()).startswith(b"\xff\xd8"))
+        finally:
+            gate.set()
+            await asyncio.wait_for(job.task, 5)
+        self.assertFalse((self.output_dir / "thumbnails").exists())
+
+    async def test_a_preview_that_cannot_be_made_serves_the_original(self):
+        broken = self.output_dir / "image.png"
+        broken.write_bytes(b"\x89PNG\r\n\x1a\n not an image")
+        start = self.capture([self.ref()])
+        descriptor = await self.media(start)
+        response = await self.current(descriptor["thumbnail_url"])
+        self.assertEqual(response.status, 200)
+        self.assertEqual(await response.read(), broken.read_bytes())
+        self.assertFalse((self.output_dir / "thumbnails" / "image_thumb.png").exists())
+
     async def test_disappearance_before_open_is_missing_and_closes_handle(
         self,
     ):

tests/test_api_images.py

@@ -80,6 +80,8 @@ class ImageAPITestCase(AioHTTPTestCase):
 
     async def tearDownAsync(self):
         await super().tearDownAsync()  # closes the aiohttp test client
+        # The automatic preview worker writes into the temp tree: stop it first.
+        self.api.close_auto_thumbnails()
         self._anchor_patch.stop()
         # Windows refuses to unlink a database that still has open handles.
         self.api.db.close_all()
@@ -1565,39 +1567,65 @@ class TestVideoThumbnailFallbacks(ImageRouteCoverageCase):
         with Image.open(self.thumb) as img:
             self.assertEqual(img.format, "JPEG")
 
-    def test_cv2_cannot_open_or_read(self):
-        with patch.dict(sys.modules, {"cv2": self._fake_cv2(opened=False)}):
-            closed = self.api._generate_video_thumbnail(
-                self.video, self.thumb, self.SIZE
-            )
-        with patch.dict(sys.modules, {"cv2": self._fake_cv2(read_ok=False)}):
-            unread = self.api._generate_video_thumbnail(
-                self.video, self.thumb, self.SIZE
-            )
-        self.assertEqual((closed, unread), (False, False))
+    def test_cv2_cannot_open_or_read_falls_through_to_ffmpeg_then_placeholder(self):
+        # #249 I1: an OpenCV that cannot decode the clip (AV1) no longer ends
+        # the run; ffmpeg and then the placeholder are tried.
+        for fake in (self._fake_cv2(opened=False), self._fake_cv2(read_ok=False)):
+            self.thumb.unlink(missing_ok=True)
+            with (
+                patch.dict(sys.modules, {"cv2": fake}),
+                patch("subprocess.Popen", side_effect=FileNotFoundError("ffmpeg")),
+            ):
+                ok = self.api._generate_video_thumbnail(
+                    self.video, self.thumb, self.SIZE
+                )
+            self.assertTrue(ok)
+            with Image.open(self.thumb) as img:
+                self.assertEqual((img.format, img.size), ("JPEG", self.SIZE))
 
     def test_cv2_crash_is_contained(self):
         cv2 = MagicMock()
         cv2.VideoCapture.side_effect = RuntimeError("driver crash")
-        with patch.dict(sys.modules, {"cv2": cv2}):
+        with (
+            patch.dict(sys.modules, {"cv2": cv2}),
+            patch("subprocess.Popen", side_effect=FileNotFoundError("ffmpeg")),
+        ):
             ok = self.api._generate_video_thumbnail(self.video, self.thumb, self.SIZE)
+        self.assertTrue(ok, "the placeholder is written")
+
+    def test_cv2_failure_keeps_an_existing_preview_on_rebuild(self):
+        make_png(self.thumb)
+        before = self.thumb.read_bytes()
+        with (
+            patch.dict(sys.modules, {"cv2": self._fake_cv2(opened=False)}),
+            patch("subprocess.Popen", side_effect=FileNotFoundError("ffmpeg")),
+        ):
+            ok = self.api._generate_video_thumbnail(
+                self.video, self.thumb, self.SIZE, keep_existing=True
+            )
         self.assertFalse(ok)
+        self.assertEqual(self.thumb.read_bytes(), before)
 
     def test_ffmpeg_absent_yields_placeholder(self):
         with (
             patch.dict(sys.modules, {"cv2": None}),
-            patch("subprocess.run", side_effect=FileNotFoundError("ffmpeg")),
+            patch("subprocess.Popen", side_effect=FileNotFoundError("ffmpeg")),
         ):
             ok = self.api._generate_video_thumbnail(self.video, self.thumb, self.SIZE)
         self.assertTrue(ok)
         with Image.open(self.thumb) as img:
             self.assertEqual(img.size, self.SIZE)
 
+    @staticmethod
+    def _process(returncode, stderr=""):
+        process = MagicMock(returncode=returncode)
+        process.communicate.return_value = ("", stderr)
+        return process
+
     def test_ffmpeg_failure_yields_placeholder(self):
-        failed = MagicMock(returncode=1, stderr="no stream")
         with (
             patch.dict(sys.modules, {"cv2": None}),
-            patch("subprocess.run", return_value=failed),
+            patch("subprocess.Popen", return_value=self._process(1, "no stream")),
         ):
             ok = self.api._generate_video_thumbnail(self.video, self.thumb, self.SIZE)
         self.assertTrue(ok)
@@ -1606,16 +1634,41 @@ class TestVideoThumbnailFallbacks(ImageRouteCoverageCase):
     def test_ffmpeg_success(self):
         with (
             patch.dict(sys.modules, {"cv2": None}),
-            patch("subprocess.run", return_value=MagicMock(returncode=0)) as run,
+            patch("subprocess.Popen", return_value=self._process(0)) as run,
         ):
             ok = self.api._generate_video_thumbnail(self.video, self.thumb, self.SIZE)
         self.assertTrue(ok)
         self.assertEqual(run.call_args.args[0][0], "ffmpeg")
 
+    def test_ffmpeg_is_tracked_and_a_hold_before_the_rename_writes_nothing(self):
+        tracked = []
+
+        def writing(cmd, **kwargs):
+            Path(cmd[-1]).write_bytes(b"\xff\xd8 frame")
+            return self._process(0)
+
+        with (
+            patch.dict(sys.modules, {"cv2": None}),
+            patch("subprocess.Popen", side_effect=writing),
+        ):
+            with self.assertRaises(images_module.PreviewWriteHeld):
+                self.api._generate_video_thumbnail(
+                    self.video,
+                    self.thumb,
+                    self.SIZE,
+                    guard=lambda: False,
+                    track=tracked.append,
+                )
+        self.assertEqual(len(tracked), 1)
+        self.assertFalse(self.thumb.exists())
+        self.assertEqual(
+            [p.name for p in self.thumb.parent.iterdir() if "part" in p.name], []
+        )
+
     def test_placeholder_failure_returns_false(self):
         with (
             patch.dict(sys.modules, {"cv2": None}),
-            patch("subprocess.run", side_effect=FileNotFoundError),
+            patch("subprocess.Popen", side_effect=FileNotFoundError),
             patch.object(
                 images_module.Image, "new", side_effect=RuntimeError("no PIL")
             ),

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_jobs.py

@@ -324,7 +324,7 @@ class TestThumbnailJob(ThumbnailJobCase):
         old = preview.read_bytes()
         with (
             patch.dict(sys.modules, {"cv2": None}),
-            patch("subprocess.run", side_effect=FileNotFoundError("no ffmpeg")),
+            patch("subprocess.Popen", side_effect=FileNotFoundError("no ffmpeg")),
         ):
             _, body = await self._start(mode="rebuild")
             job = await self._finished(body["job"]["id"])
@@ -566,7 +566,7 @@ class TestThumbnailJob(ThumbnailJobCase):
 
         with (
             patch.dict(sys.modules, {"cv2": None}),
-            patch("subprocess.run", side_effect=FileNotFoundError("no ffmpeg")),
+            patch("subprocess.Popen", side_effect=FileNotFoundError("no ffmpeg")),
             patch.object(images_module, "_atomic_save", refuse),
         ):
             _, body = await self._start()
@@ -594,12 +594,14 @@ class TestThumbnailJob(ThumbnailJobCase):
         def fake_ffmpeg(cmd, **_kwargs):
             commands.append(cmd)
             Path(cmd[-1]).write_bytes(b"jpeg")
-            return type("Done", (), {"returncode": 0, "stderr": ""})()
+            done = type("Done", (), {"returncode": 0})()
+            done.communicate = lambda timeout=None: ("", "")
+            return done
 
         for quality, qscale in (("high", "2"), ("medium", "4"), ("low", "8")):
             with (
                 patch.dict(sys.modules, {"cv2": None}),
-                patch("subprocess.run", side_effect=fake_ffmpeg),
+                patch("subprocess.Popen", side_effect=fake_ffmpeg),
             ):
                 _, body = await self._start(mode="rebuild", size=640, quality=quality)
                 job = await self._finished(body["job"]["id"])
@@ -618,12 +620,13 @@ class TestThumbnailJob(ThumbnailJobCase):
         real_save = images_module._atomic_save
 
         def recording(img, dst, fmt=None, **kwargs):
-            saved.append(kwargs)
+            # The encoder settings (guard is the automatic previews' hold check).
+            saved.append({k: v for k, v in kwargs.items() if k != "guard"})
             return real_save(img, dst, fmt, **kwargs)
 
         with (
             patch.dict(sys.modules, {"cv2": None}),
-            patch("subprocess.run", side_effect=FileNotFoundError("no ffmpeg")),
+            patch("subprocess.Popen", side_effect=FileNotFoundError("no ffmpeg")),
             patch.object(images_module, "_atomic_save", recording),
         ):
             _, body = await self._start(quality="low")

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/gallery.js

@@ -488,7 +488,8 @@
                     const index = from + offset;
                     // Use thumbnail for display if available, fallback to original
                     const displayUrl = image.thumbnail_url || image.url;
-                    const hasThumb = !!image.thumbnail_url;
+                    // The listing names a preview URL for every file (made on view, #249); the dot means the file exists.
+                    const hasThumb = typeof image.thumbnail_ready === 'boolean' ? image.thumbnail_ready : !!image.thumbnail_url;
                     const isVideo = image.is_video || false;
                     const mediaType = image.media_type || 'image';
                     
@@ -551,7 +552,8 @@
                     const index = from + offset;
                     // Use thumbnail for display if available, fallback to original
                     const displayUrl = image.thumbnail_url || image.url;
-                    const hasThumb = !!image.thumbnail_url;
+                    // The listing names a preview URL for every file (made on view, #249); the dot means the file exists.
+                    const hasThumb = typeof image.thumbnail_ready === 'boolean' ? image.thumbnail_ready : !!image.thumbnail_url;
                     const isVideo = image.is_video || false;
                     const mediaType = image.media_type || 'image';
                     

web/js/workspace/file-inspector.mjs

@@ -318,18 +318,25 @@ 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;
-        if (video) {
-            node.muted = true; node.playsInline = true; node.preload = 'metadata';
-            contentScope.cleanup(() => {node.pause(); node.removeAttribute('src'); node.load();});
-        } else {
-            node.alt = ''; node.decoding = 'async';
-            if (file?.width && file?.height) {node.width = file.width; node.height = file.height;}
-            contentScope.cleanup(() => node.removeAttribute('src'));
-        }
+        const {file} = record, preview = record.preview?.url, isVideo = file?.media_type === 'video';
+        if (isVideo && !preview) return videoCover(record);
+        const node = el('img', undefined, 'detail-cover'); node.src = preview || record.download.url;
+        node.alt = ''; node.decoding = 'async';
+        if (file?.width && file?.height) {node.width = file.width; node.height = file.height;}
+        contentScope.cleanup(() => node.removeAttribute('src'));
+        // #249: a video whose preview cannot be loaded swaps back to its own element.
+        if (isVideo) contentScope.listen(node, 'error', () => { if (node.isConnected) node.replaceWith(videoCover(record)); }, {once:true});
+        return node;
+    }
+    function videoCover(record) {
+        const node = el('video', undefined, 'detail-cover'); node.src = record.download.url;
+        node.muted = true; node.playsInline = true; node.preload = 'metadata';
+        contentScope.cleanup(() => {node.pause(); node.removeAttribute('src'); node.load();});
         return node;
     }
     /** "File & provenance": the file facts, the D21 prompt evidence, the occurrence and the saved association (D66 for a local file). */

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/image-card.mjs

@@ -17,19 +17,29 @@ function el(tag, text, className) {
     return node;
 }
 
-/** The card's media: a video's first frame when it has no thumbnail, else the thumbnail or the image. */
+/** A video's own element (its first frame), for a video without a usable preview. */
+function videoMedia(image) {
+    const node = el('video');
+    node.src = window.ApiClient.resolveUrl(image.url);
+    node.muted = true; node.playsInline = true; node.preload = 'metadata';
+    return node;
+}
+
+/**
+ * The card's media: the thumbnail (a video's is its preview frame), else the image, else a video's own element.
+ * #249: a video whose preview cannot be loaded swaps back to its own element.
+ */
 export function cardMedia(image, lazy = false) {
     const { resolveUrl } = window.ApiClient;
-    const video = image.is_video && !image.thumbnail_url;
-    const node = el(video ? 'video' : 'img');
+    if (image.is_video && !image.thumbnail_url) return videoMedia(image);
+    const node = el('img');
     node.src = resolveUrl(image.thumbnail_url || image.url);
-    if (video) { node.muted = true; node.playsInline = true; node.preload = 'metadata'; }
-    else {
-        // The concept card image: alt is the card title, with the listing's intrinsic size when known.
-        node.alt = cardTitle(image); node.decoding = 'async'; if (lazy) node.loading = 'lazy';
-        if ([image.width, image.height].every(value => Number.isFinite(value) && value > 0)) {
-            node.width = image.width; node.height = image.height;
-        }
+    // replaceWith does nothing for a card no longer in the page.
+    if (image.is_video) node.addEventListener('error', () => node.replaceWith(videoMedia(image)), { once: true });
+    // The concept card image: alt is the card title, with the listing's intrinsic size when known.
+    node.alt = cardTitle(image); node.decoding = 'async'; if (lazy) node.loading = 'lazy';
+    if ([image.width, image.height].every(value => Number.isFinite(value) && value > 0)) {
+        node.width = image.width; node.height = image.height;
     }
     return node;
 }

web/js/workspace/inspection-state.mjs

@@ -44,6 +44,9 @@ export function inspectPath(image) {
     return `/prompt_manager/images/inspect/${image.path.split('/').map(encodeURIComponent).join('/')}?${query}`;
 }
 
+/** One path segment as the server quotes it (Python `quote(..., safe="/")` also escapes `!'()*`, #249 M11). */
+const quoteSegment = segment => encodeURIComponent(segment).replace(/[!'()*]/g, c => `%${c.charCodeAt(0).toString(16).toUpperCase()}`);
+
 function downloadFor(image, required) {
     try {
         if (typeof image.url !== 'string' || !image.url) throw new TypeError('Missing original media URL');
@@ -55,7 +58,7 @@ function downloadFor(image, required) {
         const sameOrigin = 'Origin' in globalThis ?
             globalThis.Origin.from(url).isSameOrigin(globalThis.Origin.from(reference)) :
             url.origin !== 'null' && url.origin === reference.origin;
-        const expected = new URL(resolveUrl(`/prompt_manager/images/serve/${image.path.split('/').map(encodeURIComponent).join('/')}`), reference);
+        const expected = new URL(resolveUrl(`/prompt_manager/images/serve/${image.path.split('/').map(quoteSegment).join('/')}`), reference);
         const scope = descriptorScope(image);
         if (!sameOrigin || !['http:', 'https:'].includes(url.protocol) || url.username || url.password || url.hash ||
             url.pathname !== expected.pathname || url.searchParams.getAll('root').length !== 1 || url.searchParams.get('root') !== String(image.root_index) ||
@@ -68,6 +71,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(quoteSegment).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 +320,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.';