fix(thumbnails): admit previews on the loop and never send a video to an img

Final review fixes for automatic previews:
- A video preview that cannot be made is the placeholder frame (no-store),
  never the video; OpenCV failures fall through to ffmpeg and the
  placeholder. Cards and the inspector cover swap back to the video
  element when the preview image fails.
- View requests are admitted on the event loop (3 writers, one flight per
  file, at most 64 waiting; past that the fallback at once). Writes run on
  a small dedicated pool, never the default executor. The history route
  shares the admission, flight key and link checks.
- Drive, UNC, empty and dot segments are refused as text; the folder
  listing runs inside the admission.
- Holds are checked after the slot and again before the rename; only
  decode failures are remembered.
- Bounded, logged shutdown: ffmpeg is killed, no worker restarts after
  cleanup, and a reload stops the old worker.
- The listing says whether the preview file exists (thumbnail_ready) for
  the legacy gallery's dot; media in a nested thumbnails folder keeps its
  original; file names the server quotes match in the inspector.

__init__.py

@@ -143,7 +143,7 @@ try:
             ):
                 monitor.set_review_recorder(None)
                 monitor.set_thumbnail_queue(None)
-                api.close_auto_thumbnails()
+                api.close_auto_thumbnails(final=True)
 
             _review_host.on_cleanup.insert(0, _stop_review_recording)
 

py/api/history.py

@@ -737,7 +737,7 @@ class HistoryRoutesMixin:
                         resolved,
                         scope,
                         thumbnail=kind == "thumbnail",
-                        ensure_preview=self._history_preview,
+                        **media_options,
                     )
                 _check_current(scope, self, fence, epoch, generation)
                 return prepared
@@ -764,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,27 +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,
     ensure_preview=None,
+    placeholder: bytes | None = None,
 ) -> _OpenedCurrentMedia:
     """Open the current file, or its preview for ``thumbnail``.
 
-    A missing preview is made through ``ensure_preview(src, dst, is_video)``
-    (the API's bounded, single-flight automatic preview, #249) when given;
-    when it cannot be made the original is opened, as before.
+    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"
@@ -338,18 +347,14 @@ def _prepare_current_media(
         try:
             preview.stat()
         except FileNotFoundError:
-            # Never written through a preview folder linked out of the root.
-            if (
-                ensure_preview is not None
-                and _canonical(resolved.gallery_root / "thumbnails").is_relative_to(
-                    _canonical(resolved.gallery_root)
-                )
-                and ensure_preview(
-                    target, preview, target.suffix.lower() in VIDEO_EXTENSIONS
-                )
-                == "ready"
+            if ensure_preview is not None and _preview_writable(
+                resolved.gallery_root, preview
             ):
-                target = _canonical(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

@@ -263,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()
@@ -307,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:
@@ -321,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():
@@ -726,18 +788,24 @@ def media_thumbnail_rel(rel_path):
     )
 
 
-def media_thumbnail_url(output_path, rel_path, root_index, *, root_scope=None):
-    """The Images listing's thumbnail URL for *rel_path* under *output_path*
-    (one blocking stat). The Needs review queue cards and the Library use the
-    same URL (#249 F, M10).
+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=``. A missing one is
-    still listed (without ``v=``): the serve route makes it on first view and
-    falls back to the original when it cannot (py/api/thumbnail_auto.py), so
-    no card ever loads the full file because its preview was not made yet.
+    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:
         quoted = urllib.parse.quote(thumb_rel, safe="/")
         thumbnail_url = f"/prompt_manager/images/serve/{quoted}?root={root_index}"
@@ -745,7 +813,12 @@ def media_thumbnail_url(output_path, rel_path, root_index, *, root_scope=None):
         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(
@@ -763,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="/")
@@ -782,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,
@@ -2029,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.
 
@@ -2036,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,
@@ -2053,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."""
@@ -2087,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(
@@ -2235,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.
 

py/api/thumbnail_auto.py

@@ -2,21 +2,25 @@
 
 The approved mock has no repeated global thumbnail prompt
 (``docs/concepts/workspace/README.md``), so the workspace never asks. Every
-listing gives an image or video a thumbnail URL (``media_thumbnail_url``),
-and the previews behind those URLs are made here, in three ways:
+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 maps it back to its media file
-  under the same root (text containment first, then a regular file inside the
-  root, never through a link), makes the preview and serves it. At most
-  ``ON_DEMAND_LIMIT`` views write at once and one flight runs per file: a
-  second view of a preview being made waits for it. When the preview cannot be
-  made the original is served (``Cache-Control: no-store``) with its own
-  content type. A file that cannot be decoded is remembered for this session
-  (keyed by its path, size and mtime) so it is not tried again on every view,
-  and counted for the System Overview (``failed`` in
-  ``/images/thumbnails/status``). A preview another program holds open
-  (Windows) is not remembered: the next view tries again.
+  <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
@@ -38,14 +42,23 @@ touch the database, like the thumbnail job, so they run beside a scan, image
 tagging and the Needs review recording (S5 M7). They are not written while a
 job runs that replaces the database or removes media and previews: a restore,
 a duplicate removal, a sources change, the gallery's thumbnail clear or an
-``exclusive`` job (``_auto_thumbnail_blocker``). A view then gets the original
-and a queued file is left for its first view.
+``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
@@ -53,14 +66,13 @@ from aiohttp import web
 from .images import (
     GALLERY_IMAGE_EXTENSIONS,
     MEDIA_EXTENSIONS,
-    PREVIEW_IN_USE,
-    PREVIEW_NOT_WRITABLE,
     THUMBNAIL_CLEAR_TITLE,
     THUMBNAIL_ENCODINGS,
     VIDEO_EXTENSIONS,
     _has_traversal,
     _is_within,
     _thumbnail_is_fresh,
+    _video_placeholder_image,
     _worker_threads,
     media_thumbnail_rel,
 )
@@ -81,11 +93,14 @@ except ImportError:  # tests import the package top-level
 AUTO_THUMBNAIL_EDGE = DEFAULT_THUMBNAIL_EDGE
 AUTO_THUMBNAIL_SIZE = (AUTO_THUMBNAIL_EDGE, AUTO_THUMBNAIL_EDGE)
 AUTO_THUMBNAIL_ENCODING = THUMBNAIL_ENCODINGS["medium"]
-# Views that may write a preview at the same time; more wait for a slot.
+# Previews views may write at once (the size of the view pool).
 ON_DEMAND_LIMIT = 3
-# How long a view waits for a slot or another view's flight before it is
-# answered with the original (the next view tries again).
-ON_DEMAND_WAIT_SECONDS = 20
+# 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).
@@ -97,12 +112,14 @@ SCAN_THUMBNAIL_JOB_THRESHOLD = 200
 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"
-# _generate_one errors that say nothing about the file: not remembered.
-_TRANSIENT_ERRORS = frozenset(
-    {PREVIEW_IN_USE, PREVIEW_NOT_WRITABLE, "File no longer exists"}
-)
+# What the gate answers a request past MAX_WAITING.
+OVER = object()
 
 
 def _fingerprint(src):
@@ -114,10 +131,40 @@ def _fingerprint(src):
     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>``."""
-    parts = filepath.replace("\\", "/").split("/")
-    if len(parts) < 2 or parts[0] != "thumbnails" or "thumbnails" in parts[1:-1]:
+    """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 (
@@ -136,36 +183,53 @@ def _regular_file(path):
         return False
 
 
-def _thumbnail_source(root, filepath):
-    """``(src, dst, is_video)`` for the preview *filepath* under *root*, or None.
+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
 
-    *filepath* passed ``_is_preview_request`` and the serve route's traversal
-    check. The source has the preview's name without ``_thumb`` and an
-    extension that previews as the requested one (any case); a ``.jpg`` may
-    also be a video's. It must be a regular file inside *root*. ``dst`` is
-    None when the ``thumbnails`` folder lies outside *root* (a link): the
-    original is then served and nothing is written there.
+
+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)
-    parts = filepath.replace("\\", "/").split("/")
-    folder = root.joinpath(*parts[1:-1])
-    stem, extension = os.path.splitext(parts[-1])
-    base = stem[: -len("_thumb")]
-    wanted = [extension] + (sorted(VIDEO_EXTENSIONS) if extension == ".jpg" else [])
-    names = [base + ext for ext in wanted] + [base + ext.upper() for ext in wanted]
-    found = next((folder / n for n in names if _regular_file(folder / n)), None)
-    if found is None:
-        found = _scan_folder(folder, base, set(wanted))
-    if found is None or not _is_within(found, root):
-        return None
-    rel = Path(*parts[1:-1], found.name)
+    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  # a preview folder linked elsewhere is never written
+        dst = None
     return found, dst, found.suffix.lower() in VIDEO_EXTENSIONS
 
 
@@ -184,13 +248,23 @@ def _scan_folder(folder, base, extensions):
     return None
 
 
-class AutoThumbnailer:
-    """Single-flight preview writes for views, and the capture worker.
+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()
 
-    ``generate(src, dst, is_video)`` writes one preview and returns
-    ``_generate_one``'s result dict; ``may_write()`` says whether automatic
-    previews may be written now. Thread-safe: views call ``ensure`` on the
-    executor, the monitor calls ``enqueue`` from its own thread.
+
+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__(
@@ -200,7 +274,6 @@ class AutoThumbnailer:
         logger,
         *,
         workers=_worker_threads,
-        on_demand_limit=ON_DEMAND_LIMIT,
         max_queued=MAX_QUEUED,
         start_worker=True,
     ):
@@ -212,85 +285,93 @@ class AutoThumbnailer:
         self._start_worker = start_worker
         self._changed = threading.Condition()
         self._inflight = {}
-        self._followers = 0
-        self._slot_waiters = 0
-        self._batch = 0
         self._failures = collections.OrderedDict()
-        self._slots = threading.BoundedSemaphore(on_demand_limit)
         self._queue = collections.deque()
         self._queued = set()
+        self._processes = []
         self._worker = None
         self._closed = False
         self._busy = False
-
-    # ── views ─────────────────────────────────────────────────────────
+        self._batch = 0
 
     @property
     def failure_count(self):
         with self._changed:
             return len(self._failures)
 
-    def ensure(self, src, dst, is_video, *, on_demand=True):
+    @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
-        previewed, remembered), ``busy`` (no slot in time, a transient write
-        error, or another flight that did not finish) or ``deferred`` (a hold
-        refuses automatic writes). Blocking: call it off the event loop.
+        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 = os.path.normcase(str(dst))
+        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()
-            else:
-                self._followers += 1
-                self._changed.notify_all()
         if flight is not None:
-            try:
-                flight.wait(ON_DEMAND_WAIT_SECONDS)
-            finally:
-                with self._changed:
-                    self._followers -= 1
-            return self._settled(src, dst, fingerprint)
+            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, on_demand)
+            return self._lead(src, dst, is_video, fingerprint)
         finally:
             with self._changed:
                 self._inflight.pop(key, None)
                 self._changed.notify_all()
             done.set()
 
-    def _settled(self, src, dst, fingerprint):
+    def _lead(self, src, dst, is_video, fingerprint):
+        if not self._allowed():
+            return DEFERRED
         if _thumbnail_is_fresh(src, dst):
             return READY
-        with self._changed:
-            return FAILED if fingerprint in self._failures else BUSY
-
-    def _lead(self, src, dst, is_video, fingerprint, on_demand):
-        if not self._may_write():
-            return DEFERRED
-        if on_demand and not self._take_slot():
-            return BUSY
-        try:
-            if _thumbnail_is_fresh(src, dst):
-                return READY
-            result = self._generate(src, dst, is_video)
-        finally:
-            if on_demand:
-                self._slots.release()
+        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"
-        if error in _TRANSIENT_ERRORS:
-            return BUSY
         with self._changed:
             self._failures[fingerprint] = error
             while len(self._failures) > MAX_FAILURES_REMEMBERED:
@@ -301,24 +382,12 @@ class AutoThumbnailer:
         )
         return FAILED
 
-    def _take_slot(self):
-        if self._slots.acquire(blocking=False):
-            return True
-        with self._changed:
-            self._slot_waiters += 1
-            self._changed.notify_all()
-        try:
-            return self._slots.acquire(timeout=ON_DEMAND_WAIT_SECONDS)
-        finally:
-            with self._changed:
-                self._slot_waiters -= 1
-
     # ── the capture worker ────────────────────────────────────────────
 
     def enqueue(self, src, dst, is_video):
         """Queue *dst*'s preview for the worker; False when already queued,
         full or closed. Never blocks on a write."""
-        key = os.path.normcase(str(dst))
+        key = _dst_key(dst)
         with self._changed:
             if self._closed or key in self._queued:
                 return False
@@ -330,12 +399,13 @@ class AutoThumbnailer:
             if self._start_worker and (
                 self._worker is None or not self._worker.is_alive()
             ):
-                self._worker = threading.Thread(
-                    target=self._run,
-                    name="PromptManagerThumbnailWorker",
-                    daemon=True,
+                worker = threading.Thread(
+                    target=self._run, name=WORKER_NAME, daemon=True
                 )
-                self._worker.start()
+                worker.pm_preview_token = _MODULE_TOKEN
+                worker.pm_preview_close = self.close
+                self._worker = worker
+                worker.start()
             self._changed.notify_all()
         return True
 
@@ -365,22 +435,36 @@ class AutoThumbnailer:
     def _write_queued(self, item):
         _key, src, dst, is_video = item
         try:
-            self.ensure(src, dst, is_video, on_demand=False)
+            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 after its current batch; queued files are dropped."""
+        """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
-        if worker is not None and worker is not threading.current_thread():
-            worker.join(ON_DEMAND_WAIT_SECONDS)
+        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) ──────────────────────
 
@@ -392,54 +476,124 @@ class AutoThumbnailer:
         """True once the worker has nothing queued or in hand."""
         return self._wait_for(lambda: not self._busy, timeout)
 
-    def wait_followers(self, count, timeout=None):
-        """True once *count* views wait on another view's flight."""
-        return self._wait_for(lambda: self._followers >= count, timeout)
-
-    def wait_slot_waiters(self, count, timeout=None):
-        """True once *count* views wait for a write slot."""
-        return self._wait_for(lambda: self._slot_waiters >= count, timeout)
-
     def wait_batch(self, size, timeout=None):
         """True once the worker holds a batch of *size* files."""
         return self._wait_for(lambda: self._batch == size, timeout)
 
 
+class _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."""
+        """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):
-        """Stop the capture worker and forget it (shutdown, a test's reset).
+    def close_auto_thumbnails(self, final=False):
+        """Stop the capture worker and the view pool, and forget them.
 
-        Waits for the batch in hand, so nothing writes into the output tree
-        after it returns; the next use starts a fresh one.
+        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):
+    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
+            src,
+            dst,
+            AUTO_THUMBNAIL_SIZE,
+            is_video,
+            encoding=AUTO_THUMBNAIL_ENCODING,
+            stop=stop,
+            guard=guard,
+            track=track,
         )
 
     def _auto_thumbnail_blocker(self):
@@ -451,42 +605,89 @@ class AutoThumbnailMixin:
                 return job
         return None
 
-    def _history_preview(self, src, dst, is_video):
-        """A recorded run's missing preview, made like a view's (blocking)."""
-        return self._auto_thumbnails().ensure(Path(src), Path(dst), is_video)
+    # ── 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)."""
+        root_parts = (Path(dst).name,)
+        return await self._preview_gate().run(
+            flight_key(Path(dst).parent, root_parts),
+            functools.partial(self._auto_thumbnails().ensure, src, dst, is_video),
+            self._preview_pool(),
+        )
 
     # ── on first view ─────────────────────────────────────────────────
 
-    def _preview_on_demand(self, root, filepath):
-        """``(outcome, src, dst)`` for a missing preview under *root*, or None."""
-        located = _thumbnail_source(root, filepath)
+    def _preview_on_demand(self, root, parts):
+        """``(outcome, src, dst, is_video)`` for a missing preview under
+        *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, src, dst
+            return DEFERRED, src, None, is_video
         outcome = self._auto_thumbnails().ensure(src, dst, is_video)
         if outcome == READY and not dst.is_file():
-            outcome = BUSY  # removed under us: show the original this once
-        return outcome, src, dst
+            outcome = BUSY  # removed under us: the fallback this once
+        return outcome, 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 the original when it cannot be made; None
-        when *filepath* names no preview of a media file in *output_dirs*."""
+        """The made preview, or its fallback; None when *filepath* names no
+        preview of a media file in *output_dirs*."""
         if _has_traversal(filepath) or not _is_preview_request(filepath):
             return None
+        parts = tuple(filepath.split("/"))
+        gate = self._preview_gate()
+        pool = self._preview_pool()
         for root in output_dirs:
-            answer = await self._run_in_executor(
-                self._preview_on_demand, Path(root), filepath
+            root = Path(root)
+            answer = await gate.run(
+                flight_key(root, parts),
+                functools.partial(self._preview_on_demand, root, parts),
+                pool,
             )
-            if answer is None:
-                continue
-            outcome, src, dst = answer
-            if outcome == READY:
-                return self._file_response(dst)
-            response = web.FileResponse(src)
-            response.headers["Cache-Control"] = "no-store"
-            return response
+            if answer is OVER:
+                # Past the cap: exact names only, no listing, no write.
+                quick = _quick_source(root, parts)
+                answer = None if quick is None else (BUSY, quick[0], None, quick[1])
+            if answer is not None:
+                return self._preview_response(*answer)
         return None
 
     # ── on capture ────────────────────────────────────────────────────
@@ -495,9 +696,12 @@ class AutoThumbnailMixin:
         """Queue the preview of a new output (the monitor's thread).
 
         Returns ``queued``, ``fresh`` (its preview is current) or ``skipped``
-        (not a media file under a current root, a link, a preview itself, or
-        already queued). Never writes on the caller's thread.
+        (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"
@@ -521,9 +725,11 @@ class AutoThumbnailMixin:
             rel = Path(os.path.relpath(path, root))
             if "thumbnails" in rel.parts[:-1]:
                 return None
-            if not _is_within(root / "thumbnails", root):
-                return None
             dst = root / media_thumbnail_rel(rel)
+            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
 
@@ -579,3 +785,7 @@ class AutoThumbnailMixin:
         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/workspace-thumbnails.spec.mjs

@@ -104,3 +104,15 @@ test('Library cards and the shared gallery request previews only', async ({ 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);
+});

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

@@ -912,3 +912,13 @@ test('record carries the listing thumbnail as its preview, never another file (#
     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/test_api_history_outputs.py

@@ -1128,6 +1128,51 @@ class HistoryOutputAPITests(ImageAPITestCase):
         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")

tests/test_api_images.py

@@ -1567,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)
@@ -1608,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_thumbnail_auto.py

@@ -62,29 +62,53 @@ class AutoThumbnailCase(ThumbnailTestCase):
         return calls
 
     def _gate_writes(self):
-        """Hold every preview write until released: (entered, release, active)."""
+        """Hold every preview write until released: (entered, release, state).
+
+        ``state`` counts the writers in hand (``active``), their ``peak`` and
+        the ``threads`` they ran on; ``state["changed"]`` is notified at each
+        change, so a test waits for a writer's own state (never a timer).
+        """
         entered = threading.Event()
         release = threading.Event()
-        state = {"active": 0, "peak": 0}
-        lock = threading.Lock()
+        changed = threading.Condition()
+        state = {"active": 0, "peak": 0, "threads": set(), "changed": changed}
         original = self.api._generate_one
 
         def gated(*args, **kwargs):
-            with lock:
+            with changed:
                 state["active"] += 1
                 state["peak"] = max(state["peak"], state["active"])
+                state["threads"].add(threading.current_thread().name)
+                changed.notify_all()
             entered.set()
             try:
                 release.wait(WAIT)
                 return original(*args, **kwargs)
             finally:
-                with lock:
+                with changed:
                     state["active"] -= 1
+                    changed.notify_all()
 
         self.api._generate_one = gated
         self.addCleanup(release.set)
         return entered, release, state
 
+    async def _writers(self, state, count):
+        """Wait until exactly ``count`` writers are in hand."""
+        changed = state["changed"]
+
+        def wait():
+            with changed:
+                return changed.wait_for(lambda: state["active"] == count, WAIT)
+
+        loop = asyncio.get_running_loop()
+        self.assertTrue(await loop.run_in_executor(None, wait))
+
+    async def _waiting(self, count):
+        """Wait until ``count`` view requests wait on the preview gate."""
+        gate = self.api._preview_gate()
+        await asyncio.wait_for(gate.settled(lambda: gate.waiting >= count), WAIT)
+
     async def _wait(self, event):
         loop = asyncio.get_running_loop()
         self.assertTrue(await loop.run_in_executor(None, event.wait, WAIT))
@@ -130,6 +154,24 @@ class TestListingsAlwaysCarryAThumbnailUrl(AutoThumbnailCase):
             entry["thumbnail_url"], f"{SERVE}thumbnails/clip_thumb.jpg?root=0"
         )
 
+    def test_listing_says_whether_the_preview_file_exists(self):
+        # M9: the legacy gallery's dot reads thumbnail_ready.
+        make_png(self.output_dir / "a.png", size=(64, 64))
+        self.assertIs(
+            _entry(self.api, self.output_dir, "a.png")["thumbnail_ready"], False
+        )
+        make_png(self.output_dir / "thumbnails" / "a_thumb.png")
+        self.assertIs(
+            _entry(self.api, self.output_dir, "a.png")["thumbnail_ready"], True
+        )
+
+    def test_media_inside_a_nested_thumbnails_folder_keeps_its_original(self):
+        # M10: such a file is never previewed; its card shows the original.
+        make_png(self.output_dir / "project" / "thumbnails" / "x.png", size=(64, 64))
+        entry = _entry(self.api, self.output_dir, "project/thumbnails/x.png")
+        self.assertIsNone(entry["thumbnail_url"])
+        self.assertIs(entry["thumbnail_ready"], False)
+
     def test_existing_preview_keeps_its_mtime_version(self):
         make_png(self.output_dir / "a.png", size=(64, 64))
         make_png(self.output_dir / "thumbnails" / "a_thumb.png")
@@ -194,12 +236,8 @@ class TestThumbnailOnFirstView(AutoThumbnailCase):
         first = asyncio.ensure_future(self.client.get(url))
         await self._wait(entered)
         others = [asyncio.ensure_future(self.client.get(url)) for _ in range(4)]
-        # Every follower is waiting on the leader's flight before it is released.
-        thumbnailer = self.api._auto_thumbnails()
-        loop = asyncio.get_running_loop()
-        self.assertTrue(
-            await loop.run_in_executor(None, thumbnailer.wait_followers, 4, WAIT)
-        )
+        # Every follower waits on the leader's flight before it is released.
+        await self._waiting(5)
         release.set()
         responses = [await first] + [await task for task in others]
 
@@ -217,19 +255,113 @@ class TestThumbnailOnFirstView(AutoThumbnailCase):
             )
             for n in names
         ]
-        await self._wait(entered)
-        thumbnailer = self.api._auto_thumbnails()
-        loop = asyncio.get_running_loop()
-        # The extra requests queue for a slot instead of writing beside them.
-        self.assertTrue(
-            await loop.run_in_executor(None, thumbnailer.wait_slot_waiters, 2, WAIT)
-        )
+        # M8: the third writer's own state, then the two queued requests.
+        await self._writers(state, auto.ON_DEMAND_LIMIT)
+        await self._waiting(len(names))
         self.assertEqual(state["active"], auto.ON_DEMAND_LIMIT)
         release.set()
         responses = [await task for task in tasks]
         self.assertEqual([r.status for r in responses], [200] * len(names))
         self.assertLessEqual(state["peak"], auto.ON_DEMAND_LIMIT)
 
+    async def test_a_burst_never_holds_default_executor_threads(self):
+        # I2: waits are coroutines and writes run on the preview pool, so a
+        # burst of distinct missing previews leaves every other route free.
+        names = [f"b{i}.png" for i in range(auto.ON_DEMAND_LIMIT + 4)]
+        for name in names:
+            make_png(self.output_dir / name, size=(64, 64))
+        entered, release, state = self._gate_writes()
+        default_calls = []
+        original = self.api._run_in_executor
+
+        async def spy(func, *args, **kwargs):
+            default_calls.append(getattr(func, "__name__", repr(func)))
+            return await original(func, *args, **kwargs)
+
+        self.api._run_in_executor = spy
+        tasks = [
+            asyncio.ensure_future(
+                self.client.get(_entry(self.api, self.output_dir, n)["thumbnail_url"])
+            )
+            for n in names
+        ]
+        await self._writers(state, auto.ON_DEMAND_LIMIT)
+        await self._waiting(len(names))
+        self.assertEqual(default_calls, [], "no view waits on a default thread")
+        self.assertTrue(all(name.startswith("pm-preview") for name in state["threads"]))
+        status = await self.client.get("/prompt_manager/images/thumbnails/status")
+        self.assertEqual(status.status, 200, "another route answers meanwhile")
+        await status.read()
+        release.set()
+        responses = [await task for task in tasks]
+        self.assertEqual([r.status for r in responses], [200] * len(names))
+        for name in names:
+            stem = name[:-4]
+            self.assertTrue(
+                (self.output_dir / "thumbnails" / f"{stem}_thumb.png").is_file()
+            )
+
+    async def test_requests_beyond_the_cap_get_the_fallback_at_once(self):
+        names = [f"c{i}.png" for i in range(auto.ON_DEMAND_LIMIT)]
+        for name in names + ["late.png"]:
+            make_png(self.output_dir / name, size=(64, 64))
+        (self.output_dir / "late.mp4").write_bytes(b"not a video")
+        entered, release, state = self._gate_writes()
+        listings = []
+        real_scandir = auto.os.scandir
+
+        def counting(path):
+            listings.append(str(path))
+            return real_scandir(path)
+
+        with (
+            patch.object(auto, "MAX_WAITING", auto.ON_DEMAND_LIMIT),
+            patch.object(auto.os, "scandir", counting),
+        ):
+            tasks = [
+                asyncio.ensure_future(
+                    self.client.get(
+                        _entry(self.api, self.output_dir, n)["thumbnail_url"]
+                    )
+                )
+                for n in names
+            ]
+            await self._writers(state, auto.ON_DEMAND_LIMIT)
+            image = await self.client.get(f"{SERVE}thumbnails/late_thumb.png?root=0")
+            self.assertEqual(image.status, 200)
+            self.assertEqual(image.headers["Cache-Control"], "no-store")
+            self.assertEqual(
+                await image.read(), (self.output_dir / "late.png").read_bytes()
+            )
+            video = await self.client.get(f"{SERVE}thumbnails/late_thumb.jpg?root=0")
+            self.assertEqual(video.status, 200)
+            self.assertEqual(video.headers["Content-Type"], "image/jpeg")
+            self.assertEqual(video.headers["Cache-Control"], "no-store")
+            await video.read()
+            missing = await self.client.get(f"{SERVE}thumbnails/Nope_thumb.png?root=0")
+            self.assertEqual(missing.status, 404)
+            # M2: no folder is listed for a request the cap turned away.
+            self.assertEqual(listings, [])
+            release.set()
+            responses = [await task for task in tasks]
+        self.assertEqual([r.status for r in responses], [200] * len(names))
+        self.assertFalse((self.output_dir / "thumbnails" / "late_thumb.png").exists())
+
+    async def test_mixed_case_folder_listing_runs_once_inside_admission(self):
+        make_png(self.output_dir / "Shot.Png", size=(64, 64))
+        listings = []
+        real_scandir = auto.os.scandir
+
+        def counting(path):
+            listings.append(str(path))
+            return real_scandir(path)
+
+        with patch.object(auto.os, "scandir", counting):
+            resp = await self.client.get(f"{SERVE}thumbnails/Shot_thumb.png?root=0")
+        self.assertEqual(resp.status, 200)
+        self.assertEqual(len(listings), 1)
+        self.assertTrue((self.output_dir / "thumbnails" / "Shot_thumb.png").is_file())
+
     async def test_broken_file_falls_back_to_the_original_and_is_remembered(self):
         broken = self.output_dir / "broken.png"
         broken.write_bytes(b"\x89PNG\r\n\x1a\n not an image")
@@ -286,6 +418,39 @@ class TestThumbnailOnFirstView(AutoThumbnailCase):
         self.assertNotIn(str(self.output_dir), text)
         self.assertFalse((self.output_dir / "thumbnails").exists())
 
+    def test_drive_unc_and_absolute_segments_are_refused_as_text(self):
+        # M1: a drive or UNC segment would reset the anchor of a joined path.
+        for path in (
+            "thumbnails/D:/x_thumb.png",
+            "thumbnails/D:x_thumb.png",
+            "thumbnails/\\\\server\\share/x_thumb.png",
+            "thumbnails//server/share/x_thumb.png",
+            "/thumbnails/x_thumb.png",
+            "thumbnails/a/./x_thumb.png",
+            "thumbnails/C:\\Users\\x_thumb.png",
+        ):
+            with self.subTest(path=path):
+                self.assertFalse(auto._is_preview_request(path))
+
+    async def test_refused_segments_never_reach_the_filesystem(self):
+        with (
+            patch.object(auto, "_thumbnail_source") as source,
+            patch.object(auto, "_quick_source") as quick,
+            patch.object(auto.os, "stat") as stat,
+        ):
+            for path in (
+                "thumbnails/D:/x_thumb.png",
+                "thumbnails/\\\\server\\share/x_thumb.png",
+                "/thumbnails/x_thumb.png",
+            ):
+                answer = await self.api._serve_missing_thumbnail(
+                    [self.output_dir], path
+                )
+                self.assertIsNone(answer)
+            source.assert_not_called()
+            quick.assert_not_called()
+            stat.assert_not_called()
+
     async def test_traversal_is_refused_before_any_filesystem_access(self):
         with patch.object(auto, "_thumbnail_source") as source:
             for path in (
@@ -374,6 +539,188 @@ class TestThumbnailOnFirstView(AutoThumbnailCase):
         self.assertFalse((self.output_dir / "thumbnails").exists())
 
 
+class TestVideoPreviewsNeverFallBackToTheVideo(AutoThumbnailCase):
+    """I1: a video preview that cannot be made is the placeholder frame."""
+
+    def _video(self, name="clip.webm"):
+        path = self.output_dir / name
+        path.write_bytes(b"AV1 stream no decoder here")
+        return path
+
+    async def _assert_placeholder(self, url):
+        resp = await self.client.get(url)
+        self.assertEqual(resp.status, 200)
+        self.assertEqual(resp.headers["Content-Type"], "image/jpeg")
+        body = await resp.read()
+        self.assertTrue(body.startswith(b"\xff\xd8"))
+        return resp
+
+    async def test_opencv_that_cannot_open_the_file_falls_through_to_a_frame(self):
+        from unittest.mock import MagicMock
+
+        self._video()
+        cv2 = MagicMock()
+        cv2.VideoCapture.return_value.isOpened.return_value = False
+        with patch.dict(sys.modules, {"cv2": cv2}):
+            await self._assert_placeholder(f"{SERVE}thumbnails/clip_thumb.jpg?root=0")
+        self.assertTrue((self.output_dir / "thumbnails" / "clip_thumb.jpg").is_file())
+
+    async def test_a_held_video_preview_is_the_placeholder_and_writes_nothing(self):
+        from py.api.jobs import RESTORE_JOB_KIND
+
+        self._video()
+        self._hold(RESTORE_JOB_KIND)
+        resp = await self._assert_placeholder(
+            f"{SERVE}thumbnails/clip_thumb.jpg?root=0"
+        )
+        self.assertEqual(resp.headers["Cache-Control"], "no-store")
+        self.assertFalse((self.output_dir / "thumbnails").exists())
+
+    async def test_a_linked_out_folder_gives_a_video_the_placeholder(self):
+        self._video()
+        elsewhere = Path(self._output_tmp.name) / "elsewhere"
+        elsewhere.mkdir()
+        self._make_symlink(elsewhere, self.output_dir / "thumbnails")
+        resp = await self._assert_placeholder(
+            f"{SERVE}thumbnails/clip_thumb.jpg?root=0"
+        )
+        self.assertEqual(resp.headers["Cache-Control"], "no-store")
+        self.assertEqual(os.listdir(elsewhere), [])
+
+    async def test_a_failed_video_preview_is_the_placeholder(self):
+        self._video()
+        with patch.object(self.api, "_generate_video_thumbnail", return_value=False):
+            resp = await self._assert_placeholder(
+                f"{SERVE}thumbnails/clip_thumb.jpg?root=0"
+            )
+        self.assertEqual(resp.headers["Cache-Control"], "no-store")
+
+
+class TestWriteSafety(AutoThumbnailCase):
+    async def test_a_hold_that_starts_while_waiting_for_a_slot_stops_the_write(self):
+        # M3: the holds are checked again once the slot is taken.
+        from py.api.jobs import RESTORE_JOB_KIND
+
+        names = [f"h{i}.png" for i in range(auto.ON_DEMAND_LIMIT + 1)]
+        for name in names:
+            make_png(self.output_dir / name, size=(64, 64))
+        calls = self._count_writes()
+        entered, release, state = self._gate_writes()
+        tasks = [
+            asyncio.ensure_future(
+                self.client.get(_entry(self.api, self.output_dir, n)["thumbnail_url"])
+            )
+            for n in names
+        ]
+        await self._writers(state, auto.ON_DEMAND_LIMIT)
+        await self._waiting(len(names))
+        self._hold(RESTORE_JOB_KIND)
+        release.set()
+        responses = [await task for task in tasks]
+        self.assertEqual([r.status for r in responses], [200] * len(names))
+        self.assertEqual(len(calls), auto.ON_DEMAND_LIMIT, "the late one never wrote")
+        # ... and once more before the rename: the gated writers wrote nothing.
+        thumbs = self.output_dir / "thumbnails"
+        self.assertEqual(sorted(os.listdir(thumbs)) if thumbs.exists() else [], [])
+
+    async def test_disk_errors_are_not_remembered_but_decode_errors_are(self):
+        # M4: only a file that cannot be decoded is remembered.
+        import errno
+
+        make_png(self.output_dir / "full.png", size=(64, 64))
+        url = _entry(self.api, self.output_dir, "full.png")["thumbnail_url"]
+        with patch.object(
+            images_module,
+            "_write_image_thumbnail",
+            side_effect=OSError(errno.ENOSPC, "No space left on device"),
+        ):
+            resp = await self.client.get(url)
+            self.assertEqual(resp.status, 200)
+            await resp.read()
+        self.assertEqual(self.api._auto_thumbnails().failure_count, 0)
+        truncated = self.output_dir / "cut.png"
+        data = make_png(self.output_dir / "whole.png", size=(256, 256)).read_bytes()
+        truncated.write_bytes(data[: len(data) // 2])
+        resp = await self.client.get(
+            _entry(self.api, self.output_dir, "cut.png")["thumbnail_url"]
+        )
+        self.assertEqual(await resp.read(), truncated.read_bytes())
+        self.assertEqual(self.api._auto_thumbnails().failure_count, 1)
+
+
+class TestShutdown(AutoThumbnailCase):
+    def _worker_threads(self):
+        return [
+            t
+            for t in threading.enumerate()
+            if t.name == auto.WORKER_NAME and t.is_alive()
+        ]
+
+    async def test_no_thread_remains_after_cleanup_and_capture_cannot_restart(self):
+        path = make_png(self.output_dir / "a.png", size=(64, 64))
+        self.assertEqual(self.api.queue_capture_thumbnail(str(path)), "queued")
+        await self._idle()
+        resp = await self.client.get(
+            _entry(self.api, self.output_dir, "a.png")["thumbnail_url"]
+        )
+        await resp.read()
+        pool_threads = list(self.api._preview_pool_threads())
+        loop = asyncio.get_running_loop()
+        await loop.run_in_executor(
+            None, lambda: self.api.close_auto_thumbnails(final=True)
+        )
+        for thread in pool_threads:
+            await loop.run_in_executor(None, thread.join, WAIT)
+        self.assertEqual(self._worker_threads(), [])
+        self.assertFalse(any(t.is_alive() for t in pool_threads))
+        other = make_png(self.output_dir / "b.png", size=(64, 64))
+        self.assertEqual(self.api.queue_capture_thumbnail(str(other)), "skipped")
+        self.assertEqual(self._worker_threads(), [])
+
+    async def test_a_reload_stops_the_old_worker(self):
+        path = make_png(self.output_dir / "a.png", size=(64, 64))
+        self.api.queue_capture_thumbnail(str(path))
+        await self._idle()
+        (old,) = self._worker_threads()
+        old.pm_preview_token = object()  # as the module before a reload left it
+        loop = asyncio.get_running_loop()
+        await loop.run_in_executor(None, auto._stop_stale_workers)
+        await loop.run_in_executor(None, old.join, WAIT)
+        self.assertFalse(old.is_alive())
+
+    async def test_close_kills_ffmpeg_and_logs_a_join_that_timed_out(self):
+        from unittest.mock import MagicMock
+
+        entered, release, _state = self._gate_writes()
+        path = make_png(self.output_dir / "a.png", size=(64, 64))
+        self.api.queue_capture_thumbnail(str(path))
+        await self._wait(entered)
+        thumbnailer = self.api._auto_thumbnails()
+        process = MagicMock()
+        thumbnailer._track(process)
+        with (
+            patch.object(auto, "CLOSE_WAIT_SECONDS", 0),
+            self.assertLogs("prompt_manager.api", "WARNING") as logs,
+        ):
+            self.api.close_auto_thumbnails()
+        process.kill.assert_called_once()
+        self.assertIn("still running", "\n".join(logs.output))
+        release.set()
+
+
+class TestHistorySharesTheAdmission(AutoThumbnailCase):
+    def test_history_and_serve_route_share_one_flight_key(self):
+        root = self.output_dir
+        preview = root / "thumbnails" / "sub" / "a_thumb.png"
+        self.assertEqual(
+            auto.flight_key(root, ("thumbnails", "sub", "a_thumb.png")),
+            auto.flight_key(
+                Path(os.path.normcase(os.path.realpath(root))),
+                preview.relative_to(root).parts,
+            ),
+        )
+
+
 class TestThumbnailOnCapture(AutoThumbnailCase):
     async def test_monitor_output_is_queued_and_generated_in_the_background(self):
         path = make_png(self.output_dir / "run" / "ComfyUI_0001.png", size=(640, 480))

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")

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

@@ -323,16 +323,20 @@ export function mountFileInspector(container, {api, sender, parentScope, variant
      * 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, preview = record.preview?.url, video = !preview && file?.media_type === 'video';
-        const node = el(video ? 'video' : 'img', undefined, 'detail-cover'); node.src = preview || record.download.url;
-        if (video) {
-            node.muted = true; node.playsInline = true; node.preload = 'metadata';
-            contentScope.cleanup(() => {node.pause(); node.removeAttribute('src'); node.load();});
-        } 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/image-card.mjs

@@ -17,19 +17,28 @@ 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;
-        }
+    if (image.is_video) node.addEventListener('error', () => { if (node.isConnected) 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) ||
@@ -90,7 +93,7 @@ function previewFor(image) {
         const reference = new URL(location?.href || 'http://localhost/');
         const resolveUrl = globalThis.window?.ApiClient?.resolveUrl || globalThis.ApiClient?.resolveUrl || (url => url);
         const value = resolveUrl(image.thumbnail_url), url = new URL(value, reference);
-        const expected = new URL(resolveUrl(`/prompt_manager/images/serve/${previewPath(image.path).split('/').map(encodeURIComponent).join('/')}`), reference);
+        const 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) ||