feat(thumbnails): Convert outdated thumbnails as an Activity job

POST /jobs/thumbnails {mode: "convert"} starts "Convert outdated
thumbnails" under the thumbnail job's kind, guarded admission, worker
threads and cancel at a file boundary. For each file whose preview is
in another format, it writes the new preview from the source (atomic,
verified; a video with no decodable frame fails instead of getting the
placeholder), then removes that file's old-format previews in its
thumbnails folder. A failure keeps the old preview; orphan previews
and links are never touched. The final line reports converted, failed
and bytes before and after. The plan counts it with mode=convert; the
legacy gallery stream refuses to attach to a running conversion.

py/api/__init__.py

@@ -9,6 +9,7 @@ Split into domain-specific mixins for maintainability:
 - JobRoutesMixin: the session job registry behind Activity (py/api/jobs.py)
 - ImageTagJobsMixin: bulk Add tags / Auto-tag as image-tags jobs
 - ThumbnailJobsMixin: thumbnail plan, thumbnail job and the gallery stream on it
+- ThumbnailConvertMixin: Convert outdated thumbnails (#317), a thumbnail job mode
 - AutoThumbnailMixin: previews made on first view, on capture and after a scan
 - ReviewRoutesMixin: the Needs review queue, item detail and decisions
 - ReviewRecordingMixin: Needs review recording for the scan and the monitor
@@ -50,6 +51,7 @@ from .settings_routes import SettingsRoutesMixin
 from .job_routes import JobRoutesMixin
 from .image_tag_jobs import ImageTagJobsMixin
 from .thumbnail_jobs import ThumbnailJobsMixin
+from .thumbnail_convert import ThumbnailConvertMixin
 from .thumbnail_auto import AutoThumbnailMixin
 from .system import SystemRoutesMixin
 from .library_location import LibraryLocationRoutesMixin
@@ -295,6 +297,7 @@ class PromptManagerAPI(
     JobRoutesMixin,
     ImageTagJobsMixin,
     ThumbnailJobsMixin,
+    ThumbnailConvertMixin,
     AutoThumbnailMixin,
     SystemRoutesMixin,
     LibraryLocationRoutesMixin,

py/api/images.py

@@ -2134,6 +2134,7 @@ class ImageRoutesMixin:
         stop=None,
         guard=None,
         track=None,
+        placeholder=True,
     ):
         """Create one thumbnail (blocking). Never raises; returns a result dict.
 
@@ -2148,7 +2149,8 @@ class ImageRoutesMixin:
         ``track(process)`` is handed every encoder 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). A generated
-        result names the file written as ``path``.
+        result names the file written as ``path``. Without ``placeholder``
+        (Convert outdated) a video with no decodable frame fails instead.
         """
         result = {
             "file": src.name,
@@ -2174,6 +2176,7 @@ class ImageRoutesMixin:
                     keep_existing=rebuild,
                     guard=guard,
                     track=track,
+                    placeholder=placeholder,
                 )
                 if not made:
                     return {
@@ -2225,6 +2228,7 @@ class ImageRoutesMixin:
         *,
         guard=None,
         track=None,
+        placeholder=True,
     ):
         """Write a video's preview; returns the path written, or False.
 
@@ -2238,7 +2242,8 @@ class ImageRoutesMixin:
         preview propagates (``_generate_one`` names it), as do
         PreviewWriteHeld when ``guard()`` refuses the rename and
         EncoderClosed at shutdown. ``track(process)`` is handed the ffmpeg
-        process so a shutdown can kill it.
+        process so a shutdown can kill it. Without ``placeholder`` no
+        placeholder is written: the run fails when no frame decodes.
         """
         try:
             frame = self._opencv_frame(video_path, settings.size)
@@ -2258,7 +2263,7 @@ class ImageRoutesMixin:
                     f"Generated video thumbnail using {source}: {written}"
                 )
                 return written
-            if keep_existing and thumbnail_path.exists():
+            if not placeholder or (keep_existing and thumbnail_path.exists()):
                 self.logger.warning(
                     f"No frame decoded for {video_path.name}; keeping its preview"
                 )

py/api/thumbnail_convert.py

@@ -0,0 +1,247 @@
+"""Convert outdated thumbnails (#317): an explicit, cancellable Activity job.
+
+``POST /prompt_manager/jobs/thumbnails {mode: "convert"}`` (and
+``GET /prompt_manager/images/thumbnails/plan?mode=convert`` for its count)
+starts "Convert outdated thumbnails": the thumbnail job's kind, ``guarded``
+admission (never beside a record-changing job such as a scan, a restore or
+a duplicate removal; one thumbnail job at a time), its worker threads and
+its outcome groups. Nothing converts on its own: changing the format in
+Appearance only makes previews "outdated".
+
+A preview is outdated when the file owns one in another format than the one
+written now (``writer_settings``: a legacy ``_thumb.png``, or a format
+switched away from). For each such file, at a file boundary (Cancel keeps
+what was converted):
+
+1. the new preview is written from the SOURCE media in the configured
+   format (``_generate_one``: atomic, verified to decode; a video whose frame
+   cannot be decoded fails rather than getting the placeholder), unless a
+   current preview at least as new as the source is already there;
+2. only then are that file's old-format previews removed: regular files (no
+   links) in its ``thumbnails/`` folder, owned by that source
+   (``thumbnail_settings.owns``), with the source still present.
+
+A failure keeps the old preview. Previews whose source is gone (orphans) are
+never planned, so never touched. The final status line reads
+"N converted · F failed · <bytes before> → <bytes after>".
+"""
+
+import asyncio
+import functools
+import os
+import time
+from pathlib import Path
+
+from . import thumbnail_settings
+from .admin import _plural, _scan_location, _scan_roots, _scan_status_line
+from .images import _thumbnail_is_fresh, _worker_threads, _thumbnail_batch_size
+from .prompts import safe_error_message
+
+try:
+    from ...utils.parallel import map_parallel
+except ImportError:  # tests import the package top-level
+    from utils.parallel import map_parallel
+
+CONVERT_JOB_TITLE = "Convert outdated thumbnails"
+NOTHING_OUTDATED = "No outdated previews to convert"
+CANCELLED_BEFORE_CONVERTING = "Cancelled before any preview was converted"
+SOURCE_GONE = "The file is gone; its old preview was kept"
+OLD_NOT_REMOVED = "Converted; the old preview could not be removed"
+_UNITS = ("B", "KB", "MB", "GB", "TB")
+
+
+def readable_bytes(count):
+    """``1.5 KB``-style size (whole bytes below 1 KB)."""
+    size = float(count)
+    for unit in _UNITS:
+        if size < 1024 or unit == _UNITS[-1]:
+            return f"{int(size)} B" if unit == "B" else f"{size:.1f} {unit}"
+        size /= 1024
+    return f"{size:.1f} {_UNITS[-1]}"  # pragma: no cover - loop always returns
+
+
+def _old_previews(src, dst, settings):
+    """``[(path, size)]``: *src*'s previews beside *dst* in another format."""
+    stem = dst.name[: -len(dst.suffix)] if dst.suffix else dst.name
+    old = []
+    for path in thumbnail_settings.owned_in_folder(
+        dst.parent, stem, src.suffix, settings.extension
+    ):
+        if path.suffix == settings.extension:
+            continue
+        try:
+            old.append((path, path.stat().st_size))
+        except OSError:
+            continue
+    return old
+
+
+def _finished_line(totals, remaining):
+    line = (
+        f"{totals['converted']:,} converted · {totals['failed']:,} failed · "
+        f"{readable_bytes(totals['before'])} → {readable_bytes(totals['after'])}"
+    )
+    if remaining:
+        line += f" · {remaining:,} more on the next run"
+    return line
+
+
+class ThumbnailConvertMixin:
+    """The Convert outdated job (started through ``_start_thumbnail_job``)."""
+
+    async def _run_convert_job(self, job, options):
+        try:
+            state = await self._convert_job_body(job, options)
+        except Exception:
+            self.logger.exception("Thumbnail conversion failed")
+            job.update(status_line="Thumbnail conversion stopped on an internal error.")
+            job.log_line("Conversion failed · internal error")
+            state = "failed"
+        job.finish(state)
+
+    async def _convert_job_body(self, job, options):
+        output_dirs = await self._run_in_executor(self._get_all_output_dirs)
+        if not output_dirs:
+            message = "ComfyUI output directory not found"
+            job.update(status_line=message)
+            job.log_line(f"Conversion failed · {message}")
+            return "failed"
+        folders = _plural(len(output_dirs), "output folder")
+        job.scope["output_folders"] = len(output_dirs)
+        job.update(status_line=f"Looking for outdated previews in {folders}")
+        job.log_line(
+            f"Conversion started · {folders} · to {options.settings.describe()}"
+        )
+        plan = await self._run_in_executor(
+            self._plan_gallery_thumbnails,
+            output_dirs,
+            "convert",
+            stop=lambda: job.cancelled,
+            settings=options.settings,
+        )
+        if job.cancelled:
+            return self._end_conversion(job, 0, CANCELLED_BEFORE_CONVERTING)
+        targets = plan["targets"]
+        if not targets:
+            job.update(progress=100, status_line=NOTHING_OUTDATED)
+            job.log_line(f"Conversion finished · {NOTHING_OUTDATED.lower()}")
+            return "finished"
+        totals = await self._convert_targets(job, options, targets, output_dirs)
+        self.invalidate_gallery_cache()
+        if job.cancelled:
+            return self._end_conversion(
+                job,
+                totals["converted"],
+                f"Cancelled · {_plural(totals['converted'], 'preview')} converted",
+            )
+        line = _finished_line(totals, plan["remaining"])
+        job.update(progress=100, status_line=line)
+        job.log_line(f"Conversion finished · {line}")
+        self.logger.info(f"Thumbnail conversion: {line}")
+        return "finished"
+
+    @staticmethod
+    def _end_conversion(job, converted, status_line):
+        job.update(status_line=status_line)
+        job.log_line(
+            f"Conversion cancelled · {_plural(converted, 'preview')} converted"
+        )
+        return "cancelled"
+
+    async def _convert_targets(self, job, options, targets, output_dirs):
+        """Convert *targets* in worker batches, recording each file."""
+        roots = _scan_roots(output_dirs)
+        total = len(targets)
+        job.update(total=total, status_line=_scan_status_line(0, total, 0, 0))
+        totals = {"converted": 0, "failed": 0, "before": 0, "after": 0}
+        start = time.monotonic()
+        batch_size = _thumbnail_batch_size(_worker_threads())
+        index = batches = 0
+        for batch_start in range(0, total, batch_size):
+            if job.cancelled:
+                break
+            batch = targets[batch_start : batch_start + batch_size]
+            results = await self._run_in_executor(
+                self._convert_batch, batch, options.settings, lambda: job.cancelled
+            )
+            for (src, _dst, _video), result in zip(batch, results):
+                if result["action"] == "cancelled":
+                    continue
+                index += 1
+                self._record_conversion(job, src, result, roots, totals)
+                job.update(processed=index, progress=index * 100 // total)
+            batches += 1
+            job.update(
+                status_line=_scan_status_line(
+                    index, total, batches, time.monotonic() - start
+                )
+            )
+            await asyncio.sleep(0)  # let other requests run between batches
+        return totals
+
+    @staticmethod
+    def _record_conversion(job, src, result, roots, totals):
+        path, label, index = _scan_location(src, roots)
+        where = f"{label}/{path}" if label else path
+        row = {"path": path, "root": label, "root_index": index}
+        if result["action"] == "converted":
+            totals["converted"] += 1
+            totals["before"] += result["before"]
+            totals["after"] += result["after"]
+            detail = result["detail"]
+            job.record("generated", {**row, "detail": detail}, f"+ {where} · {detail}")
+            return
+        totals["failed"] += 1
+        detail = result.get("error") or "failed"
+        job.record("errors", {**row, "detail": detail}, f"! {where} · {detail}")
+
+    def _convert_batch(self, targets, settings, stop):
+        """Convert ``targets`` on the configured worker threads (blocking)."""
+        convert = functools.partial(self._convert_target, settings=settings, stop=stop)
+        return map_parallel(convert, targets, _worker_threads())
+
+    def _convert_target(self, target, settings, stop):
+        src, dst, is_video = target
+        return self._convert_one(src, dst, is_video, settings, stop=stop)
+
+    def _convert_one(self, src, dst, is_video, settings, stop=None):
+        """Convert one file's preview (blocking; never raises).
+
+        ``converted`` with the bytes before and after, ``error`` (the old
+        preview kept) or ``cancelled`` (a file boundary).
+        """
+        if stop is not None and stop():
+            return {"action": "cancelled"}
+        dst = Path(dst)
+        try:
+            old = _old_previews(src, dst, settings)
+            if not _thumbnail_is_fresh(src, dst):
+                made = self._generate_one(
+                    src, dst, settings, is_video, rebuild=True, placeholder=False
+                )
+                if made["action"] != "generated":
+                    return {"action": "error", "error": made.get("error")}
+                dst = Path(made["path"])
+            if dst.suffix != settings.extension or not dst.is_file():
+                return {"action": "error", "error": "The new preview was not written"}
+            if not os.path.isfile(src):
+                return {"action": "error", "error": SOURCE_GONE}
+            kept = []
+            for path, _size in old:
+                try:
+                    os.remove(path)
+                except OSError as exc:
+                    self.logger.warning(f"Could not remove old preview {path}: {exc}")
+                    kept.append(path)
+            if kept:
+                return {"action": "error", "error": OLD_NOT_REMOVED}
+            label = thumbnail_settings.FORMAT_LABELS[settings.format]
+            return {
+                "action": "converted",
+                "before": sum(size for _path, size in old),
+                "after": dst.stat().st_size,
+                "detail": f"Converted to {label}",
+            }
+        except Exception as exc:
+            self.logger.warning(f"Conversion failed for {src.name}", exc_info=True)
+            return {"action": "error", "error": safe_error_message(exc)}

py/api/thumbnail_jobs.py

@@ -10,8 +10,9 @@
   Its outcome groups are ``generated``, ``skipped`` (already fresh) and
   ``errors``.
 
-``mode`` is ``missing`` (previews that are absent or older than their file)
-or ``rebuild`` (every preview again, fresh ones too). Format, quality and
+``mode`` is ``missing`` (previews that are absent or older than their file),
+``rebuild`` (every preview again, fresh ones too) or ``convert`` (Convert
+outdated: previews in another format, see ``thumbnail_convert.py``). Format, quality and
 size limit are the server's thumbnail settings (``thumbnail_settings``,
 #317), read once when the job starts and shown in its scope; a ``size`` or
 ``quality`` an older client sends is ignored. Every preview is written next
@@ -49,12 +50,13 @@ from .images import (
 )
 from .job_routes import busy_response
 from .jobs import THUMBNAIL_GROUP_NAMES, THUMBNAILS_JOB_KIND, JobBusy
+from .thumbnail_convert import CONVERT_JOB_TITLE
 from .thumbnail_encode import writer_settings
 from .thumbnail_settings import ThumbnailSettings
 
 THUMBNAIL_JOB_TITLE = "Thumbnail generation"
-THUMBNAIL_MODES = ("missing", "rebuild")
-_OPTIONS_ERROR = 'mode must be "missing" or "rebuild"'
+THUMBNAIL_MODES = ("missing", "rebuild", "convert")
+_OPTIONS_ERROR = 'mode must be "missing", "rebuild" or "convert"'
 CANCELLED_ERROR = "Thumbnail generation was cancelled"
 CANCELLED_BEFORE_WRITING = "Cancelled before any preview was written"
 NOTHING_TO_PREVIEW = "No images or videos to preview"
@@ -187,14 +189,20 @@ class ThumbnailJobsMixin:
         return web.json_response({"success": True, "job": job.summary()}, status=202)
 
     def _start_thumbnail_job(self, options):
-        """Register the thumbnail job synchronously; raises JobBusy."""
+        """Register the thumbnail job synchronously; raises JobBusy.
+
+        ``convert`` runs Convert outdated (``thumbnail_convert.py``) under the
+        same kind, admission, cancel and groups.
+        """
+        convert = options.mode == "convert"
+        runner = self._run_convert_job if convert else self._run_thumbnail_job
         return self._jobs.start(
             THUMBNAILS_JOB_KIND,
-            THUMBNAIL_JOB_TITLE,
+            CONVERT_JOB_TITLE if convert else THUMBNAIL_JOB_TITLE,
             admission="guarded",
             cancellable=True,
             scope=options.scope(),
-            runner=functools.partial(self._run_thumbnail_job, options=options),
+            runner=functools.partial(runner, options=options),
             groups=THUMBNAIL_GROUP_NAMES,
         )
 
@@ -455,6 +463,8 @@ class ThumbnailJobsMixin:
         the way.
         """
         running = self._jobs.active(THUMBNAILS_JOB_KIND)
+        if running is not None and running.scope.get("mode") == "convert":
+            return None, False, running  # no legacy frames to follow (#317)
         if running is not None and running.cancellable:
             return running, False, None
         busy = self._jobs.blocker(THUMBNAILS_JOB_KIND, admission="guarded")

py/api/thumbnail_settings.py

@@ -161,18 +161,19 @@ def _is_preview_file(path):
         return False
 
 
-def _owned(root, rel_path, preferred):
-    rel_path = Path(rel_path)
-    stem = Path(root) / thumbnail_stem_rel(rel_path)
-    extensions = list(candidate_extensions(rel_path.suffix, preferred))
-    if rel_path.suffix != rel_path.suffix.lower() and owns(
-        rel_path.suffix, rel_path.suffix
+def owned_in_folder(folder, stem, source_extension, preferred=None):
+    """Yield the existing previews ``<folder>/<stem><ext>`` a source with
+    *source_extension* owns, in lookup order (regular files, no links; one
+    file found under two spellings on a case-insensitive disk once)."""
+    extensions = list(candidate_extensions(source_extension, preferred))
+    if source_extension != source_extension.lower() and owns(
+        source_extension, source_extension
     ):
         # An older release kept the source's own spelling (``b_thumb.JPG``).
-        extensions.append(rel_path.suffix)
+        extensions.append(source_extension)
     seen = set()
     for extension in extensions:
-        path = stem.with_name(stem.name + extension)
+        path = Path(folder) / (stem + extension)
         if not _is_preview_file(path):
             continue
         try:
@@ -186,6 +187,12 @@ def _owned(root, rel_path, preferred):
         yield path
 
 
+def _owned(root, rel_path, preferred):
+    rel_path = Path(rel_path)
+    stem = Path(root) / thumbnail_stem_rel(rel_path)
+    return owned_in_folder(stem.parent, stem.name, rel_path.suffix, preferred)
+
+
 def owned_thumbnails(root, rel_path, preferred=None):
     """Every existing preview of *rel_path* under *root*, in lookup order."""
     return list(_owned(root, rel_path, preferred))

tests/test_thumbnail_convert.py

@@ -0,0 +1,281 @@
+"""Convert outdated thumbnails (#317): an explicit, cancellable Activity job.
+
+``POST /prompt_manager/jobs/thumbnails {mode: "convert"}`` runs under the
+thumbnail job's kind, admission (``guarded``) and worker threads. For every
+media file whose preview is in another format than the one written now, it
+writes the new preview from the SOURCE (verified, atomic), then removes that
+file's old-format previews under ``thumbnails/``. A failure keeps the old
+preview; previews whose source is gone (orphans) are never touched. The final
+line reports converted / failed and bytes before → after.
+
+Ordering is driven by events, never by timers.
+"""
+
+import asyncio
+import os
+import sys
+import threading
+import unittest
+from unittest.mock import patch
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+try:
+    from tests.test_api_images import ThumbnailTestCase, make_png
+except ImportError:  # discovered with tests/ as the top-level directory
+    from test_api_images import ThumbnailTestCase, make_png
+
+from PIL import Image  # noqa: E402
+
+from py.api import thumbnail_encode  # noqa: E402
+from py.config import PromptManagerConfig  # noqa: E402
+
+JOB = "/prompt_manager/jobs/thumbnails"
+PLAN = "/prompt_manager/images/thumbnails/plan"
+STATUS = "/prompt_manager/images/thumbnails/status"
+WAIT = 10
+
+
+def big_png(path, size=(512, 512)):
+    """A noisy PNG: its PNG preview is large, a WebP one much smaller."""
+    path.parent.mkdir(parents=True, exist_ok=True)
+    Image.effect_noise(size, 40).convert("RGB").save(path, "PNG")
+    return path
+
+
+class ConvertCase(ThumbnailTestCase):
+    async def asyncSetUp(self):
+        await super().asyncSetUp()
+        saved = PromptManagerConfig.get_config()["thumbnails"]
+        self.addCleanup(PromptManagerConfig.update_config, {"thumbnails": saved})
+        PromptManagerConfig.update_config(
+            {"thumbnails": {"format": "webp", "quality": 80, "edge": 512}}
+        )
+        thumbnail_encode.reset_encoders()
+        self.addCleanup(thumbnail_encode.reset_encoders)
+        self.addCleanup(self.api.close_auto_thumbnails)
+
+    def _legacy(self, name, suffix=".png", source=True):
+        """A media file (unless ``source`` is False) and its old-format preview."""
+        if source:
+            big_png(self.output_dir / f"{name}.png")
+        preview = big_png(self._thumb_path(name, suffix), size=(256, 256))
+        if source:
+            later = (self.output_dir / f"{name}.png").stat().st_mtime + 5
+            os.utime(preview, (later, later))
+        return preview
+
+    async def _convert(self):
+        resp = await self.client.post(JOB, json={"mode": "convert"})
+        body = await resp.json()
+        self.assertEqual(resp.status, 202, body)
+        job = self.api._jobs.get(body["job"]["id"])
+        await asyncio.wait_for(job.task, WAIT)
+        return body["job"], job
+
+
+class TestConvertOutdated(ConvertCase):
+    async def test_plan_counts_the_outdated_previews(self):
+        self._legacy("a")
+        self._legacy("b")
+        make_png(self.output_dir / "new.png")  # missing: not this job's work
+        plan = await (await self.client.get(PLAN, params={"mode": "convert"})).json()
+        self.assertEqual(
+            (plan["mode"], plan["targets"], plan["outdated"]), ("convert", 2, 2)
+        )
+        self.assertEqual(self.api._jobs.list(), [])
+
+    async def test_converts_from_the_source_and_removes_the_old_preview(self):
+        old = self._legacy("a")
+        original = (self.output_dir / "a.png").read_bytes()
+        orphan = self._legacy("gone", source=False)
+        make_png(self.output_dir / "new.png")
+        before = old.stat().st_size
+
+        summary, job = await self._convert()
+
+        self.assertEqual(
+            (summary["kind"], summary["admission"], summary["cancellable"]),
+            ("thumbnails", "guarded", True),
+        )
+        self.assertEqual(summary["title"], "Convert outdated thumbnails")
+        self.assertEqual(summary["scope"]["mode"], "convert")
+        self.assertEqual(job.state, "finished")
+        self.assertEqual(job.counters["generated"], 1)
+        self.assertEqual(job.counters["errors"], 0)
+        new = self._thumb_path("a")
+        with Image.open(new) as img:
+            self.assertEqual((img.format, max(img.size)), ("WEBP", 512))
+        self.assertFalse(old.exists(), "the old PNG preview is removed")
+        self.assertEqual((self.output_dir / "a.png").read_bytes(), original)
+        self.assertTrue(orphan.exists(), "an orphan preview is never touched")
+        self.assertFalse(self._thumb_path("new").exists(), "missing is not converted")
+        after = new.stat().st_size
+        self.assertRegex(job.status_line, r"^1 converted · 0 failed · .+ → .+$")
+        self.assertTrue(any("Conversion finished" in line for line in job.log))
+        self.assertLess(after, before)
+        status = await (await self.client.get(STATUS)).json()
+        self.assertEqual((status["outdated"], status["pending"]), (0, 1))
+
+    async def test_a_source_that_fails_keeps_its_old_preview(self):
+        old = self._legacy("bad")
+        (self.output_dir / "bad.png").write_bytes(b"\x89PNG broken")
+        later = old.stat().st_mtime + 10
+        os.utime(old, (later, later))
+        keep = old.read_bytes()
+
+        _, job = await self._convert()
+
+        self.assertEqual(job.counters["errors"], 1)
+        self.assertEqual(job.counters["generated"], 0)
+        self.assertEqual(old.read_bytes(), keep)
+        self.assertFalse(self._thumb_path("bad").exists())
+        self.assertRegex(job.status_line, r"^0 converted · 1 failed")
+
+    async def test_a_video_without_a_frame_keeps_its_old_preview(self):
+        (self.output_dir / "clip.mp4").write_bytes(b"no decoder")
+        old = big_png(self._thumb_path("clip", ".jpg"), size=(64, 64))
+        keep = old.read_bytes()
+        with (
+            patch.object(self.api, "_opencv_frame", return_value=None),
+            patch.object(self.api, "_ffmpeg_frame", return_value=None),
+        ):
+            _, job = await self._convert()
+        self.assertEqual(job.counters["errors"], 1)
+        self.assertEqual(old.read_bytes(), keep)
+        self.assertFalse(self._thumb_path("clip").exists(), "never the placeholder")
+
+    async def test_an_old_preview_that_is_a_link_is_left(self):
+        big_png(self.output_dir / "l.png")
+        target = big_png(self.comfy_dir / "elsewhere.png", size=(32, 32))
+        link = self._thumb_path("l", ".png")
+        link.parent.mkdir(parents=True, exist_ok=True)
+        self._make_symlink(target, link)
+        _, job = await self._convert()
+        self.assertTrue(link.is_symlink())
+        self.assertTrue(target.exists())
+        self.assertEqual(job.counters["errors"], 0)
+
+    async def test_a_preview_already_current_only_loses_its_leftovers(self):
+        old = self._legacy("c")
+        current = big_png(self._thumb_path("c"), size=(64, 64))
+        later = old.stat().st_mtime + 5
+        os.utime(current, (later, later))
+        written = current.read_bytes()
+        _, job = await self._convert()
+        self.assertEqual(job.counters["generated"], 1)
+        self.assertEqual(current.read_bytes(), written, "not written again")
+        self.assertFalse(old.exists())
+
+    async def test_cancel_stops_at_a_file_boundary_keeping_what_converted(self):
+        original = PromptManagerConfig.WORKER_THREADS
+        PromptManagerConfig.WORKER_THREADS = 1
+        self.addCleanup(setattr, PromptManagerConfig, "WORKER_THREADS", original)
+        olds = [self._legacy(f"p{i}") for i in range(6)]
+        done = []
+        real = self.api._convert_one
+        job_ids = []
+
+        def counting(*args, **kwargs):
+            result = real(*args, **kwargs)
+            done.append(result)
+            if len(done) == 2:
+                self.assertTrue(self.api._jobs.cancel(job_ids[0]))
+            return result
+
+        self.api._convert_one = counting
+        resp = await self.client.post(JOB, json={"mode": "convert"})
+        job_ids.append((await resp.json())["job"]["id"])
+        job = self.api._jobs.get(job_ids[0])
+        await asyncio.wait_for(job.task, WAIT)
+
+        self.assertEqual(job.state, "cancelled")
+        self.assertEqual(job.counters["generated"], 2)
+        converted = [p for p in olds if not p.exists()]
+        self.assertEqual(len(converted), 2)
+        for old in olds:
+            webp = old.with_suffix(".webp")
+            self.assertEqual(webp.exists(), not old.exists(), old.name)
+        self.assertTrue(any("Conversion cancelled" in line for line in job.log))
+
+    async def test_refused_beside_a_running_thumbnail_job_or_scan(self):
+        self._legacy("a")
+        entered, release = threading.Event(), threading.Event()
+        self.addCleanup(release.set)
+        real = self.api._convert_one
+
+        def gated(*args, **kwargs):
+            entered.set()
+            release.wait(WAIT)
+            return real(*args, **kwargs)
+
+        self.api._convert_one = gated
+        resp = await self.client.post(JOB, json={"mode": "convert"})
+        first = (await resp.json())["job"]["id"]
+        loop = asyncio.get_running_loop()
+        self.assertTrue(await loop.run_in_executor(None, entered.wait, WAIT))
+        again = await self.client.post(JOB, json={"mode": "missing"})
+        self.assertEqual(again.status, 409)
+        scan = await self.client.post("/prompt_manager/jobs/scan")
+        self.assertEqual(scan.status, 409)
+        release.set()
+        await asyncio.wait_for(self.api._jobs.get(first).task, WAIT)
+
+    async def test_the_legacy_gallery_stream_never_attaches_to_a_conversion(self):
+        try:
+            from tests.test_api_images import parse_sse
+        except ImportError:
+            from test_api_images import parse_sse
+
+        self._legacy("a")
+        entered, release = threading.Event(), threading.Event()
+        self.addCleanup(release.set)
+        real = self.api._convert_one
+
+        def gated(*args, **kwargs):
+            entered.set()
+            release.wait(WAIT)
+            return real(*args, **kwargs)
+
+        self.api._convert_one = gated
+        resp = await self.client.post(JOB, json={"mode": "convert"})
+        first = (await resp.json())["job"]["id"]
+        loop = asyncio.get_running_loop()
+        self.assertTrue(await loop.run_in_executor(None, entered.wait, WAIT))
+        stream = await self.client.post(self.PROGRESS_URL)
+        events = parse_sse(await stream.text())
+        self.assertEqual(events[-1][0], "error")
+        self.assertIn("Convert outdated thumbnails", events[-1][1]["error"])
+        release.set()
+        await asyncio.wait_for(self.api._jobs.get(first).task, WAIT)
+
+    async def test_nothing_outdated_says_so(self):
+        big_png(self.output_dir / "a.png")
+        _, job = await self._convert()
+        self.assertEqual(job.state, "finished")
+        self.assertEqual(job.status_line, "No outdated previews to convert")
+
+    async def test_the_source_vanishing_keeps_the_old_preview(self):
+        old = self._legacy("v")
+        real = self.api._generate_one
+
+        def vanish(src, *args, **kwargs):
+            os.remove(src)
+            return real(src, *args, **kwargs)
+
+        self.api._generate_one = vanish
+        _, job = await self._convert()
+        self.assertEqual(job.counters["errors"], 1)
+        self.assertTrue(old.exists(), "a source gone mid-run makes an orphan")
+
+    def test_byte_sizes_are_readable(self):
+        from py.api.thumbnail_convert import readable_bytes
+
+        self.assertEqual(readable_bytes(0), "0 B")
+        self.assertEqual(readable_bytes(1536), "1.5 KB")
+        self.assertEqual(readable_bytes(5 * 1024 * 1024), "5.0 MB")
+        self.assertEqual(readable_bytes(3 * 1024**3), "3.0 GB")
+
+
+if __name__ == "__main__":
+    unittest.main()