Merge remote-tracking branch 'origin/main' into fix/369-threshold-step

* fix(threads): stop background work at exit, on cleanup and on the tests' unload (#301)

- __init__.py: _stop_background_work() detaches the monitor's callbacks,
  stops the monitor, closes the preview worker and shuts the prompt
  cleanup thread down against one 10 s deadline, once. It runs from
  atexit (ComfyUI never runs aiohttp's cleanup hooks), from on_cleanup
  where a host runs it, and from the tests' package unload; it removes
  its own atexit callback. Registered after the history hook, so at exit
  it runs before the history dispose.
- AutoThumbnailer.close/close_auto_thumbnails take a timeout and return
  whether the worker ended.
- Tests: teardown result, once-only, on_cleanup, atexit order, a failing
  step or a second Ctrl+C never skipping the tracker, one shared
  deadline, three repeated loads leaving no thread, and a child process
  that exits normally with the image worker alive.

* fix(threads): refuse new previews before the monitor stops; isolate the exit child (#301)

Review findings:
- The teardown makes the API refuse new previews before it detaches the
  monitor's callbacks: an image in hand that already read the preview
  queue no longer starts a preview worker during exit (Python 3.12+
  refuses that thread and logs a traceback). New
  refuse_auto_thumbnails().
- The teardown imports its bounds inside each step and falls back to the
  standard logger, so a failing import never skips the other steps.
- The exit child keeps its library and log in its own folder
  (PROMPT_MANAGER_DATA_DIR), reports watchdog's threads too, fails with
  the child's output on a hang, and rejects "Error while stopping".
- TESTING.md: "Background threads at exit".

TESTING.md

@@ -275,6 +275,72 @@ Start fixing as soon as a failed log shows the cause, but push the fix only afte
 the run for that commit has finished: pull-request runs use `cancel-in-progress`,
 so a push cancels the running jobs and can hide a second failure.
 
+## Background threads at exit
+
+PromptManager's own background threads stop through one package teardown,
+`_stop_background_work()` in `__init__.py` (#301, #302). It runs once and
+works against one 10 s deadline. Each step gets the smaller of its own bound
+and the time left. The steps, in order:
+
+1. Detach the monitor's callbacks (Needs review recorder, preview queue,
+   Images listing listener).
+2. `ImageMonitor.stop_monitoring` closes the image handler, which joins the
+   image worker (5 s). It then calls watchdog's `observer.stop()`, which is
+   not bounded here, and joins the observer (5 s).
+3. `close_auto_thumbnails(final=True)` stops the preview worker (30 s) and
+   kills ffmpeg.
+4. `PromptTracker.shutdown` stops the prompt cleanup thread (2 s).
+
+A failing step, or a second Ctrl+C, is logged with its traceback and does not
+skip the later steps. A clean run logs one INFO line to PromptManager's log:
+`Background work stopped in 0.4 s: monitor=ok thumbnails=ok tracker=ok`. If
+any part is still running, one WARNING goes to the console:
+`Background work not stopped in time (10.0 s): monitor`.
+
+ComfyUI never runs aiohttp's `on_cleanup` hooks, because it never calls
+`runner.cleanup()`. So `atexit` is the live host's path. It runs on Ctrl+C
+(`main.py` catches `KeyboardInterrupt` and returns) and on a normal exit.
+The teardown is registered after the execution history hook, so at exit
+(newest first) it runs before the history writer's 2 s dispose.
+It also runs from `on_cleanup` on hosts that do run it (our test servers) and
+from the tests' package unload (`cleanup_package`), and it removes its own
+`atexit` callback. Nothing runs when the process is killed: SIGTERM, a
+ComfyUI-Manager restart through `os.execv`, a closed Windows console, or
+ComfyUI Desktop stopping its server. SQLite's WAL journal covers those.
+
+Stop contract: `PromptTracker.shutdown(timeout)`,
+`ImageGenerationHandler.shutdown(timeout)` (restartable),
+`ImageGenerationHandler.close(timeout)` (final: `schedule()` refuses and a
+sealed run's re-check is dropped) and `ImageMonitor.stop_monitoring(timeout)`
+each signal their thread and join it within the bound. They return True only
+when it ended, and log a warning naming the part when it did not. Called on
+their own thread, they only signal.
+
+Test rules:
+
+- Every test that constructs a real `PromptTracker`, `ImageGenerationHandler`
+  or `ImageMonitor` ends with its final stop and an assertion
+  (`tests/background_stops.py`: `close_handler`, `stop_monitor`,
+  `stop_tracker`). The stop comes before the folder and databases are removed.
+  `tearDown` runs before `addCleanup` callbacks, so a class whose `tearDown`
+  removes files stops the thread there.
+- A test waits for a state through an `Event` set by the code under test, or
+  by a wrapper around it, with a 30 s ceiling. It never sleeps or polls for
+  that state.
+- A test that asserts "still running" first waits for an `entered` event
+  inside the held call, then stops with `timeout=0.05`.
+- A test that asserts a history writer's `dispose()`/`close()` is True right
+  after queuing work drains first (`drain_writer`), or holds the write itself
+  (#370).
+- Patch the module attribute of the code under test (for example
+  `utils.prompt_tracker.time`, or the package's `_clock`), or patch with
+  `wraps=`. Never replace a `time` or `atexit` function for the whole
+  process.
+- `tests/test_package_exit.py` runs a child process that imports the
+  package, leaves the image worker alive and returns. Its `atexit` reporter
+  checks the teardown's result. This is the interpreter path ComfyUI takes
+  after Ctrl+C, on all three OS.
+
 ## Performance budgets
 
 A spec performance budget gets two tests:
@@ -3521,7 +3587,8 @@ npm run --prefix tools/vendor test:workflows -- workspace-system-integrations.sp
 System › Auto-tagging (`web/js/workspace/system-tagging.mjs`, D132–D134) edits one
 server-side tagging configuration, `autotag.defaults` in config.json: `model_type`
 (`wd14-swinv2`, `wd14-vit`, `gguf`, `hf`), `existing` (`skip`, `merge`,
-`replace_after_review`), `general_threshold` and `character_threshold` (0–1),
+`replace_after_review`), `general_threshold` and `character_threshold` (0–1 in
+steps of 0.05, #369),
 `template` (non-blank, at most 4,000 characters and 16 KiB) and `suggestion_limit`
 (1–100). They are host drafts (`autotag_model_type`, `autotag_existing`,
 `autotag_general_threshold`, `autotag_character_threshold`, `autotag_template`,
@@ -3532,7 +3599,13 @@ naming the setting by its row label ("General-tag confidence must be between 0 a
 `settingRefusal` in system.mjs only for a save that carried `autotag_*` keys; every
 other section's plain 400 keeps the generic status) and the control is highlighted.
 The template's limit counts code points, as Python's `len` does. A hand-edited bad value
-falls back to its default on load, one key at a time. `GET /autotag/models` reports the
+falls back to its default on load, one key at a time. Every page shows the thresholds in
+steps of 0.05 (the System rows, the Tag image dialogs, the legacy sliders), so a saved
+threshold off that grid is refused ("General-tag confidence must be a multiple of
+0.05.", `validate_saved_threshold`; `draftProblem` holds back Preview tagging review and
+the batch Review with the same sentence) and falls back on load; an accepted one is
+stored exactly on the grid (#369). A per-request threshold keeps its own check: `/image-annotations/suggest` and
+`/jobs/autotag` take 0–1; the legacy `/autotag/single` and `/autotag/start` only require a number. `GET /autotag/models` reports the
 saved values (`wd14_general_threshold`, `wd14_character_threshold`, `suggestion_limit`,
 `default_prompt`, `model_type`, `existing`), so the Tag image and Auto-tag selected
 dialogs start from them; a preset (the bulk review, Preview tagging review) wins.

__init__.py

@@ -22,7 +22,10 @@ Features:
 - Comprehensive logging and diagnostics
 """
 
+import atexit
 import re
+import threading
+import time
 from pathlib import Path
 
 
@@ -72,6 +75,12 @@ NODE_DISPLAY_NAME_MAPPINGS = {
 # are served by our own routes instead (see py/api).
 WEB_DIRECTORY = "web/comfy"
 
+# Parts the package teardown stops (``_stop_background_work``). Each stays None
+# when its start-up block below fails, and the teardown skips it.
+api = None
+_global_prompt_tracker = None
+_global_image_monitor = None
+
 # Add API routes (same pattern as ComfyUI_Assets)
 try:
     # Set extension URI
@@ -89,8 +98,6 @@ try:
     api.add_routes(routes)
     # Temp files of preview writes a crash interrupted (#317), off the
     # import path: the walk may cover a large thumbnails folder.
-    import threading
-
     threading.Thread(
         target=api.sweep_thumbnail_temps,
         name="PromptManagerPreviewTempSweep",
@@ -141,29 +148,16 @@ try:
 
     # The Needs review pass for ambiguous new outputs runs through the API's
     # capture store (the one the scan and the Tag dialog use): a linked output
-    # is never queued and its pending items leave the queue. It stops before
-    # that store is closed at shutdown.
-    _review_api = globals().get("api")
-    if _review_api is not None:
-        _global_image_monitor.set_review_recorder(_review_api.record_review_file)
+    # is never queued and its pending items leave the queue. The package
+    # teardown detaches it before that store is closed at shutdown.
+    if api is not None:
+        _global_image_monitor.set_review_recorder(api.record_review_file)
         # Every settled output's preview is queued for the API's background
         # worker (#249): the workspace never asks to generate thumbnails.
-        _global_image_monitor.set_thumbnail_queue(_review_api.queue_capture_thumbnail)
+        _global_image_monitor.set_thumbnail_queue(api.queue_capture_thumbnail)
         # Every settled output walks its Images listing again (#348): the
         # API hops from the monitor's thread onto the server's loop.
-        _global_image_monitor.set_output_listener(_review_api.notify_output_settled)
-        _review_host = getattr(config.server_instance, "app", None)
-        if _review_host is not None and hasattr(_review_host, "on_cleanup"):
-
-            async def _stop_review_recording(
-                app, monitor=_global_image_monitor, api=_review_api
-            ):
-                monitor.set_review_recorder(None)
-                monitor.set_thumbnail_queue(None)
-                monitor.set_output_listener(None)
-                api.close_auto_thumbnails(final=True)
-
-            _review_host.on_cleanup.insert(0, _stop_review_recording)
+        _global_image_monitor.set_output_listener(api.notify_output_settled)
 
     # Start monitoring
     _global_image_monitor.start_monitoring()
@@ -233,6 +227,144 @@ except Exception:
             "Failed to register execution history"
         )
 
+
+# ── Package teardown (#301, #302) ────────────────────────────────────────────
+# ComfyUI never runs aiohttp's cleanup hooks (it never calls
+# ``runner.cleanup()``), so ``atexit`` is the live host's path: Ctrl+C, and a
+# normal exit. It is registered after the history hook, so at exit (newest
+# first) it runs before the history dispose. Hosts that do run ``on_cleanup``
+# (our test servers) run it there, and the tests' package unload calls it.
+
+TEARDOWN_SECONDS = 10.0
+_clock = time.monotonic
+_background_work_lock = threading.Lock()
+_background_work_result = None
+
+
+def _teardown_step(name, stop, logger):
+    """Run one teardown step; a failure (even Ctrl+C) is logged and is False."""
+    try:
+        return bool(stop())
+    except (Exception, KeyboardInterrupt):
+        logger.error(f"Background work: stopping {name} failed", exc_info=True)
+        return False
+
+
+def _teardown_logger():
+    """PromptManager's init logger; the standard one if that cannot load."""
+    try:
+        from .utils.logging_config import get_logger
+
+        return get_logger("prompt_manager.init")
+    except Exception:
+        import logging
+
+        return logging.getLogger("prompt_manager.init")
+
+
+def _stop_background_work():
+    """Stop the image monitor, the preview worker and the prompt cleanup
+    thread against one TEARDOWN_SECONDS deadline, once.
+
+    Returns ``{"monitor", "thumbnails", "tracker"}`` booleans (True: it
+    stopped), leaving out parts that never started. Later calls return a
+    copy of the first result.
+    """
+    global _background_work_result
+    with _background_work_lock:
+        if _background_work_result is not None:
+            return dict(_background_work_result)
+        logger = _teardown_logger()
+        began = _clock()
+        deadline = began + TEARDOWN_SECONDS
+
+        def left(own_bound):
+            return max(0.0, min(own_bound, deadline - _clock()))
+
+        monitor, thumbnails, tracker = (
+            _global_image_monitor,
+            api,
+            _global_prompt_tracker,
+        )
+        result = {}
+        if monitor is not None:
+
+            def stop_monitor():
+                from .utils.image_monitor import OBSERVER_JOIN_SECONDS
+
+                # Callbacks first: an image in hand queues no preview into a
+                # closing preview worker. The worker may have read the queue
+                # callback already, so the API refuses new previews too (at
+                # exit, Python 3.12+ refuses to start their worker thread).
+                if thumbnails is not None:
+                    thumbnails.refuse_auto_thumbnails()
+                monitor.set_review_recorder(None)
+                monitor.set_thumbnail_queue(None)
+                monitor.set_output_listener(None)
+                return monitor.stop_monitoring(left(OBSERVER_JOIN_SECONDS))
+
+            result["monitor"] = _teardown_step(
+                "the image monitor", stop_monitor, logger
+            )
+        if thumbnails is not None:
+
+            def close_previews():
+                from .py.api.thumbnail_auto import CLOSE_WAIT_SECONDS
+
+                return thumbnails.close_auto_thumbnails(
+                    final=True, timeout=left(CLOSE_WAIT_SECONDS)
+                )
+
+            result["thumbnails"] = _teardown_step(
+                "the preview worker", close_previews, logger
+            )
+        if tracker is not None:
+
+            def stop_tracker():
+                from .utils.prompt_tracker import TRACKER_JOIN_SECONDS
+
+                return tracker.shutdown(left(TRACKER_JOIN_SECONDS))
+
+            result["tracker"] = _teardown_step(
+                "the prompt cleanup thread", stop_tracker, logger
+            )
+        elapsed = _clock() - began
+        failed = [name for name, ok in result.items() if not ok]
+        if failed:
+            logger.warning(
+                f"Background work not stopped in time ({elapsed:.1f} s): "
+                + ", ".join(failed)
+            )
+        else:
+            parts = " ".join(f"{name}=ok" for name in result)
+            logger.info(f"Background work stopped in {elapsed:.1f} s: {parts}")
+        _background_work_result = result
+        atexit.unregister(_stop_background_work)
+        return dict(result)
+
+
+async def _stop_background_work_on_cleanup(app):
+    """aiohttp ``on_cleanup``: no request is served then, so it runs inline."""
+    _stop_background_work()
+
+
+try:
+    atexit.register(_stop_background_work)
+    from .py.config import server_instance as _teardown_server
+
+    _teardown_host = getattr(_teardown_server, "app", None)
+    if _teardown_host is not None and hasattr(_teardown_host, "on_cleanup"):
+        _teardown_host.on_cleanup.insert(0, _stop_background_work_on_cleanup)
+except Exception as e:
+    try:
+        from .utils.logging_config import get_logger
+
+        get_logger("prompt_manager.init").error(
+            f"Failed to register the background work teardown: {e}"
+        )
+    except Exception:
+        pass
+
 __all__ = ["NODE_CLASS_MAPPINGS", "NODE_DISPLAY_NAME_MAPPINGS", "WEB_DIRECTORY"]
 
 # Print startup message with loaded tools

py/api/settings_routes.py

@@ -173,8 +173,8 @@ def _validate_autotag_settings(data):
         EXISTING_POLICIES,
         MAX_SUGGESTION_LIMIT,
         MODEL_TYPES,
+        validate_saved_threshold,
         validate_template,
-        validate_threshold,
     )
 
     updates = {}
@@ -198,7 +198,7 @@ def _validate_autotag_settings(data):
     ):
         if key in data:
             try:
-                value = validate_threshold(label, data[key])
+                value = validate_saved_threshold(label, data[key])
             except ValueError as error:
                 raise ValueError(f"{error}.") from None
             updates[key.upper()] = value

py/api/thumbnail_auto.py

@@ -488,10 +488,16 @@ class AutoThumbnailer:
                 f"Automatic preview failed for {src.name}", exc_info=True
             )
 
-    def close(self):
+    def close(self, timeout=None):
         """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."""
+        dropped), kill a running ffmpeg and wait up to ``timeout`` seconds
+        (None: CLOSE_WAIT_SECONDS, read now); a worker still running after
+        that is logged.
+
+        Returns True when there was no worker or it ended; False when it is
+        still running, or when called on the worker itself (it only signals).
+        """
+        wait = CLOSE_WAIT_SECONDS if timeout is None else timeout
         with self._changed:
             self._closed = True
             self._queue.clear()
@@ -504,14 +510,18 @@ class AutoThumbnailer:
                 process.kill()
             except OSError:
                 pass
-        if worker is None or worker is threading.current_thread():
-            return
-        worker.join(CLOSE_WAIT_SECONDS)
+        if worker is None:
+            return True
+        if worker is threading.current_thread():
+            return False
+        worker.join(wait)
         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"
+                f"{wait} s; it stops at its next file"
             )
+            return False
+        return True
 
     # ── waits (tests and shutdown; never timers) ──────────────────────
 
@@ -609,12 +619,19 @@ class AutoThumbnailMixin:
                 )
             return self._auto_thumbnailer
 
-    def close_auto_thumbnails(self, final=False):
+    def refuse_auto_thumbnails(self):
+        """Refuse every later capture without stopping the worker yet (the
+        package teardown, before ``close_auto_thumbnails(final=True)``)."""
+        with AutoThumbnailMixin._auto_thumbnailer_lock:
+            self._auto_thumbnails_shut = True
+
+    def close_auto_thumbnails(self, final=False, timeout=None):
         """Stop the capture worker and the view pool, and forget them.
 
-        Bounded (see AutoThumbnailer.close). ``final`` (the app's cleanup)
-        also keeps a late monitor event from starting a worker again; a
-        test reset passes False and the next use starts fresh ones.
+        Bounded by ``timeout`` (see AutoThumbnailer.close). ``final`` (the
+        package teardown) also keeps a late monitor event from starting a
+        worker again; a test reset passes False and the next use starts
+        fresh ones. Returns True when no capture worker is left running.
         """
         with AutoThumbnailMixin._auto_thumbnailer_lock:
             if final:
@@ -622,12 +639,14 @@ class AutoThumbnailMixin:
             thumbnailer, self._auto_thumbnailer = self._auto_thumbnailer, None
             pool, self._preview_pool_obj = self._preview_pool_obj, None
             self._preview_gate_obj = None
+        closed = True
         if thumbnailer is not None:
-            thumbnailer.close()
+            closed = thumbnailer.close(timeout)
         if pool is not None:
             pool.shutdown(wait=False, cancel_futures=True)
         if final:
             shutdown_encoders()  # kill JPEG XL / ffmpeg encoders still running
+        return closed
 
     def _auto_thumbnail_failures(self):
         thumbnailer = self._auto_thumbnailer

py/config.py

@@ -573,6 +573,7 @@ try:
         MAX_INSTRUCTION_CHARS,
         MODEL_TYPES,
         WD14_MODEL_TYPES,
+        validate_saved_threshold,
         validate_template,
         validate_threshold,
     )
@@ -585,6 +586,7 @@ except ImportError:
         MAX_INSTRUCTION_CHARS,
         MODEL_TYPES,
         WD14_MODEL_TYPES,
+        validate_saved_threshold,
         validate_template,
         validate_threshold,
     )
@@ -631,12 +633,12 @@ class AutotagConfig:
         "existing": ("EXISTING", _valid_existing, "skip"),
         "general_threshold": (
             "GENERAL_THRESHOLD",
-            lambda v: validate_threshold("value", v),
+            lambda v: validate_saved_threshold("value", v),
             0.35,
         ),
         "character_threshold": (
             "CHARACTER_THRESHOLD",
-            lambda v: validate_threshold("value", v),
+            lambda v: validate_saved_threshold("value", v),
             0.85,
         ),
         "template": ("TEMPLATE", validate_template, DEFAULT_TAG_TEMPLATE),

py/tagging_defaults.py

@@ -39,6 +39,25 @@ def validate_threshold(name: str, value: Any) -> float:
     return float(value)
 
 
+# Every page shows the thresholds in steps of 0.05 (the concept's System rows,
+# the Tag image dialogs and the legacy sliders), so a saved one must be on that
+# grid (#369). Per-request thresholds keep their own checks.
+THRESHOLD_STEP = 0.05
+_STEPS_PER_UNIT = round(1 / THRESHOLD_STEP)
+# A computed float can miss the grid by a hair: (0.1 + 0.2) * 20 is 6.000000000000001.
+_STEP_TOLERANCE = 1e-12  # in steps; refuses 0.35000000001
+
+
+def validate_saved_threshold(name: str, value: Any) -> float:
+    """A saved WD14 threshold: in [0, 1] and a multiple of ``THRESHOLD_STEP``,
+    returned exactly on the grid; else ValueError naming ``name``."""
+    value = validate_threshold(name, value)
+    steps = value * _STEPS_PER_UNIT
+    if abs(steps - round(steps)) > _STEP_TOLERANCE:
+        raise ValueError(f"{name} must be a multiple of {THRESHOLD_STEP}")
+    return round(steps) / _STEPS_PER_UNIT
+
+
 def validate_template(value: Any, name: str = "instructions") -> str:
     """Non-empty language-model instructions within the character and byte
     caps, or ValueError naming ``name``."""

tests/background_stops.py

@@ -0,0 +1,27 @@
+"""Final stops for the background threads a test starts (#301, #302).
+
+Every real ``ImageGenerationHandler``, ``ImageMonitor`` or ``PromptTracker``
+a test constructs ends with its final stop and an assertion, before the
+test's folder and databases go: ``close()`` for a handler,
+``stop_monitoring()`` for a monitor, ``shutdown()`` for a tracker.
+"""
+
+from utils.prompt_tracker import PromptTracker
+
+
+def stop_tracker(testcase, tracker):
+    """Shut a real tracker's cleanup thread down; mocks are left alone."""
+    if isinstance(tracker, PromptTracker):
+        testcase.assertTrue(tracker.shutdown(), "prompt cleanup thread still running")
+
+
+def close_handler(testcase, handler):
+    """Close ``handler`` for good, then stop the tracker it was built with."""
+    testcase.assertTrue(handler.close(), "image worker still running")
+    stop_tracker(testcase, getattr(handler, "prompt_tracker", None))
+
+
+def stop_monitor(testcase, monitor):
+    """Stop ``monitor`` (its observer and worker), then its tracker."""
+    testcase.assertTrue(monitor.stop_monitoring(), "image monitor still running")
+    stop_tracker(testcase, getattr(monitor, "prompt_tracker", None))

tests/browser/workspace-system-tagging.spec.mjs

@@ -188,6 +188,35 @@ test('a refused value shows the server\'s message', async ({ page, request }) =>
     await expect(actionOf(page, 'Preview tagging review')).not.toHaveAttribute('aria-disabled', 'true');
 });
 
+test('a value off the 0.05 step is refused and never reaches the Tag image dialog (#369)', async ({ page, request }) => {
+    const before = (await serverJson(request, SETTINGS_URL)).settings;
+    const STEP = 'General-tag confidence must be a multiple of 0.05.';
+    await selectThenOpen(page, FIRST);
+    await page.locator('#threshold-general').fill('0.37');
+    await page.locator('#threshold-general').press('Tab');
+    // Preview tagging review explains instead of opening a dialog whose Generate the browser would block.
+    await expect(actionOf(page, 'Preview tagging review')).toHaveAttribute('aria-disabled', 'true');
+    await actionOf(page, 'Preview tagging review').focus();
+    await page.keyboard.press('Enter');
+    await expect(toast(page)).toHaveText(STEP);
+    await expect(tagDialog(page)).toHaveCount(0);
+    // The server refuses the save with the same sentence and writes nothing.
+    await save(page).click();
+    await expect(status(page)).toHaveText(STEP);
+    expect((await serverJson(request, SETTINGS_URL)).settings.autotag_general_threshold).toBe(before.autotag_general_threshold);
+    // A value on the step opens the dialog, and its Generate passes the browser's own check and sends.
+    await page.locator('#threshold-general').fill('0.4');
+    await page.locator('#threshold-general').press('Tab');
+    await expect(actionOf(page, 'Preview tagging review')).not.toHaveAttribute('aria-disabled', 'true');
+    await actionOf(page, 'Preview tagging review').click();
+    await expect(tagDialog(page).locator('#image-general-threshold')).toHaveValue('0.4');
+    expect(await tagDialog(page).locator('#image-general-threshold').evaluate(input => input.checkValidity())).toBe(true);
+    const suggested = page.waitForResponse(item => item.url().endsWith('/image-annotations/suggest'));
+    await tagDialog(page).getByRole('button', { name: 'Generate suggestions', exact: true }).click();
+    const answer = await suggested;
+    expect(answer.request().postDataJSON().options.general_threshold).toBe(0.4);
+});
+
 test('Download starts the job and Activity shows it without Cancel', async ({ page, request }) => {
     await unavailable(request, ['wd14-vit']);
     await fakeDownload(request, 'hold');

tests/browser/workspace_system_fixture.py

@@ -97,15 +97,25 @@ def register_system_commands(app, routes, api, root: Path):
     real_move = library_move.move_library
     keep = PromptManagerConfig.KEEP_LIBRARY_IN_NODE_FOLDER
     config_bytes = config_path.read_bytes() if config_path.exists() else None
-    host_values = (
-        PromptManagerConfig.RESULT_TIMEOUT,
-        PromptManagerConfig.WEBUI_DISPLAY_MODE,
-        PromptManagerConfig.WORKER_THREADS,
-        GalleryConfig.MONITORING_ENABLED,
-        PromptManagerConfig.THUMBNAIL_FORMAT,
-        PromptManagerConfig.THUMBNAIL_QUALITY,
-        PromptManagerConfig.THUMBNAIL_EDGE,
+    from py.api.settings_routes import (
+        _AUTOTAG_SETTINGS,
+        _GALLERY_SETTINGS,
+        _SETTINGS_FILE_KEYS,
     )
+
+    # Every value POST /settings can change, so a reset restores each one
+    # (a saved safety_backup=false once leaked into a later spec's review).
+    # The tagging settings are the tagging fixture's to reset.
+    host_values = {
+        (GalleryConfig if name in _GALLERY_SETTINGS else PromptManagerConfig, name): (
+            getattr(
+                GalleryConfig if name in _GALLERY_SETTINGS else PromptManagerConfig,
+                name,
+            )
+        )
+        for name in _SETTINGS_FILE_KEYS
+        if name not in _AUTOTAG_SETTINGS
+    }
     from py.api import thumbnail_encode
 
     monitor = getattr(image_monitor, "_monitor_instance", None)
@@ -135,15 +145,8 @@ def register_system_commands(app, routes, api, root: Path):
             unplugged.rmdir()
         GalleryConfig.MONITORING_DIRECTORIES = list(configured)
         api._health_report = None
-        (
-            PromptManagerConfig.RESULT_TIMEOUT,
-            PromptManagerConfig.WEBUI_DISPLAY_MODE,
-            PromptManagerConfig.WORKER_THREADS,
-            GalleryConfig.MONITORING_ENABLED,
-            PromptManagerConfig.THUMBNAIL_FORMAT,
-            PromptManagerConfig.THUMBNAIL_QUALITY,
-            PromptManagerConfig.THUMBNAIL_EDGE,
-        ) = host_values
+        for (owner, name), value in host_values.items():
+            setattr(owner, name, value)
         thumbnail_encode.reset_encoders()
         image_monitor._monitor_instance = monitor
         api.__dict__.pop("_library_resolver", None)

tests/comfyui/fixtures/ledger_nodes.py

@@ -1,5 +1,6 @@
 """Deterministic nodes and queue controls installed only in an owned test host."""
 
+import asyncio
 import os
 import threading
 import time
@@ -291,7 +292,8 @@ async def image_monitor(request: web.Request) -> web.Response:
         config.MONITORING_ENABLED = True
         monitor.start_monitoring([folder_paths.get_output_directory()])
     else:
-        monitor.stop_monitoring()
+        # Joins the image worker and the observer (up to 5 s each): off the loop.
+        await asyncio.to_thread(monitor.stop_monitoring)
         config.MONITORING_ENABLED = False
     return web.json_response({"monitoring": monitor.is_monitoring})
 

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

@@ -307,6 +307,16 @@ test('Preview tagging review opens the Tag image dialog with the section’s val
     assert.deepEqual(draftProblem({ ...HOST, autotag_general_threshold: 1.5 }),
         { key: 'autotag_general_threshold', message: 'General-tag confidence must be between 0 and 1.' });
     assert.equal(draftProblem({ ...HOST, autotag_character_threshold: '' }).key, 'autotag_character_threshold');
+    // #369: every page shows the thresholds in steps of 0.05, so a value off that grid cannot be saved or previewed.
+    assert.deepEqual(draftProblem({ ...HOST, autotag_general_threshold: 0.37 }),
+        { key: 'autotag_general_threshold', message: 'General-tag confidence must be a multiple of 0.05.' });
+    assert.deepEqual(draftProblem({ ...HOST, autotag_character_threshold: 0.33 }),
+        { key: 'autotag_character_threshold', message: 'Character-tag confidence must be a multiple of 0.05.' });
+    assert.equal(draftProblem({ ...HOST, autotag_general_threshold: 0.15, autotag_character_threshold: 0.95 }), null);
+    assert.equal(draftProblem({ ...HOST, autotag_general_threshold: 0, autotag_character_threshold: 1 }), null);
+    // A computed value a hair off the grid passes and is handed on snapped to it, as the server stores it.
+    assert.equal(draftProblem({ ...HOST, autotag_general_threshold: 0.1 + 0.2 }), null);
+    assert.equal(taggingPreset({ ...HOST, autotag_general_threshold: 0.1 + 0.2 }).options.general_threshold, 0.3);
     assert.equal(draftProblem({ ...HOST, autotag_suggestion_limit: 2.5 }).key, 'autotag_suggestion_limit');
     assert.equal(draftProblem({ ...HOST, autotag_suggestion_limit: 0 }, { limit: false }), null);
     assert.deepEqual(draftProblem({ ...HOST, autotag_model_type: 'hf', autotag_template: '  ' }),
@@ -342,6 +352,11 @@ test('Preview tagging review opens the Tag image dialog with the section’s val
         await button('Preview tagging review').dispatch('click');
         assert.deepEqual(calls.toasts, ['General-tag confidence must be between 0 and 1.']);
         assert.equal(calls.opened.length, 1);
+        byId('threshold-general').value = '0.37'; await byId('threshold-general').dispatch('change');
+        assert.equal(button('Preview tagging review').getAttribute('aria-disabled'), 'true');
+        await button('Preview tagging review').dispatch('click');
+        assert.deepEqual(calls.toasts.at(-1), 'General-tag confidence must be a multiple of 0.05.');
+        assert.equal(calls.opened.length, 1);
     } finally { scope.dispose(); dom.restore(); }
 });
 
@@ -361,6 +376,8 @@ test('the batch review grid, body and Cancel-only reasons (D134, CUJ 4–5)', as
     assert.equal(existingLabel('odd'), 'odd');
     assert.deepEqual(batchBody(HOST), { model_type: 'wd14-swinv2', existing: 'skip',
         options: { general_threshold: 0.35, character_threshold: 0.85 } });
+    // #369: a threshold a hair off the grid is sent snapped to it.
+    assert.equal(batchBody({ ...HOST, autotag_character_threshold: 0.1 + 0.2 }).options.character_threshold, 0.3);
     assert.deepEqual(batchBody(llm), { model_type: 'gguf', existing: 'skip', options: { instructions: HOST.autotag_template } });
     assert.equal(batchCloseNote(PLAN), null);
     assert.equal(batchCloseNote({ ...PLAN, review_required: true }), REPLACE_NOTE);

tests/test_config.py

@@ -516,8 +516,9 @@ class TestAutotagConfig(unittest.TestCase):
         bad_values = {
             "model_type": ["wd14-convnext", 3, None],
             "existing": ["replace", "", True],
-            "general_threshold": [1.5, -0.1, float("nan"), "0.4", True],
-            "character_threshold": [2, float("inf"), None],
+            # #369: a saved threshold off the 0.05 step no page can show.
+            "general_threshold": [1.5, -0.1, float("nan"), "0.4", True, 0.37],
+            "character_threshold": [2, float("inf"), None, 0.33],
             # 4000 characters are at most 16,000 UTF-8 bytes, so the 16 KiB
             # byte cap can only be met past the character cap.
             "template": ["", "   ", "x" * 4001, "\U0001f600" * 4097, "\ud800", 7],
@@ -567,6 +568,27 @@ class TestAutotagConfig(unittest.TestCase):
             config_module.MODEL_TYPES, {"gguf", "hf", "wd14-swinv2", "wd14-vit"}
         )
 
+    def test_saved_threshold_is_a_multiple_of_the_step(self):
+        """#369: every page shows the thresholds in steps of 0.05, so a saved
+        one must be on that grid; it is stored exactly on it."""
+        from py.tagging_defaults import validate_saved_threshold as validate
+
+        for value in (0, 0.05, 0.15, 0.35, 0.85, 1):
+            with self.subTest(value=value):
+                self.assertEqual(validate("x", value), value)
+        self.assertEqual(validate("x", 0.1 + 0.2), 0.3)
+        for value, message in (
+            (0.37, "x must be a multiple of 0.05"),
+            (0.351, "x must be a multiple of 0.05"),
+            (0.35000000001, "x must be a multiple of 0.05"),
+            (1.5, "x must be between 0 and 1"),
+            ("0.4", "x must be a finite number"),
+        ):
+            with self.subTest(value=value):
+                with self.assertRaises(ValueError) as caught:
+                    validate("x", value)
+                self.assertEqual(str(caught.exception), message)
+
     def test_suggestion_limit_setting_validated_on_load(self):
         for value in (0, 101, "10", float("nan"), 2.5, True, None, [10]):
             with self.subTest(value=value):

tests/test_execution_history_hooks.py

@@ -28,6 +28,8 @@ except ImportError:
 
 Status = namedtuple("ExecutionStatus", "status_str completed messages")
 JOB = "a0c26da3-eb26-405c-b143-c662a84c3112"
+# A cross-thread wait or a duration check fails only past this (#370).
+CEILING_SECONDS = 30.0
 
 
 class HostQueue:
@@ -209,6 +211,9 @@ class HistoryHooksTests(unittest.TestCase):
         self.queue.get = unrelated_after
         self.dequeue(0)
         self.queue.task_done(0, {}, status())
+        # The writer drains first: this test checks wrappers and the session,
+        # not the 2 s bound, which a slow runner can spend on the write (#370).
+        self.drain(first)
         self.assertTrue(first.dispose())
         self.assertTrue(first.dispose())
         self.assertIs(self.queue.get, unrelated_after)
@@ -339,15 +344,17 @@ class HistoryHooksTests(unittest.TestCase):
         with patch.object(self.db, "write_execution_start", blocked):
             registration = self.register()
             self.dequeue(0)
-            self.assertTrue(entered.wait(1))
+            # Ceilings, not tight wall clocks (#370): the return values show
+            # the bound; a ceiling still fails a call that waits unbounded.
+            self.assertTrue(entered.wait(CEILING_SECONDS))
             began = time.monotonic()
             self.queue.task_done(0, {}, status())
             self.dequeue(1)
-            self.assertLess(time.monotonic() - began, 0.2)
+            self.assertLess(time.monotonic() - began, CEILING_SECONDS)
             began = time.monotonic()
             self.assertFalse(registration.dispose())
             self.assertGreaterEqual(time.monotonic() - began, 1.8)
-            self.assertLess(time.monotonic() - began, 2.3)
+            self.assertLess(time.monotonic() - began, CEILING_SECONDS)
             release.set()
             self.assertTrue(registration.dispose(2))
 

tests/test_execution_history_writer.py

@@ -642,6 +642,8 @@ class HistoryWriterTests(unittest.TestCase):
                 == "succeeded"
             )
             self.assertTrue(writer.submit(self.start_command(2)))
+            # Drain first: the 2 s bound is not under test here (#370).
+            drain_writer(writer)
             self.assertTrue(writer.close(2))
         self.assertEqual(len(set(threads)), 1)
         self.assertNotEqual(threads[0], threading.get_ident())

tests/test_execution_output_hooks.py

@@ -245,6 +245,8 @@ class OutputHooksTests(fixtures.unittest.TestCase):
         registration = self.register()
         self.dequeue(0)
         late = registration._task_done
+        # Drain first: the 2 s bound is not under test here (#370).
+        self.drain(registration)
         self.assertTrue(registration.dispose())
         thread = threading.Thread(
             target=lambda: late((0, history(), fixtures.status()), {})

tests/test_execution_output_writer.py

@@ -17,6 +17,7 @@ from utils.execution_contexts import MAX_ATTEMPT_BYTES
 from utils.execution_output_capture import seal_outputs
 from utils.execution_outputs import snapshot_output_roots
 from utils.execution_history_writer import HistoryWriter
+from tests.history_waits import drain_writer
 
 try:
     from tests import test_execution_history_writer as fixtures
@@ -161,6 +162,8 @@ class OutputWriterTests(fixtures.unittest.TestCase):
         start = replace(self.start_command(), run_id=str(fixtures.uuid4()))
         self.assertTrue(writer.submit(start))
         self.finalization(writer, start)
+        # Drain first: the 2 s bound is not under test here (#370).
+        drain_writer(writer)
         self.assertTrue(writer.close(2))
         self.assertEqual(
             self.db.get_execution_run(start.run_id)["outcome"], "succeeded"

tests/test_hashing.py

@@ -12,6 +12,7 @@ import unittest
 
 sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
 
+from tests.background_stops import stop_tracker  # noqa: E402
 from utils.hashing import (
     generate_prompt_hash,
     is_duplicate_prompt,
@@ -49,6 +50,7 @@ class TestSingleHashImplementation(unittest.TestCase):
         db = mock.Mock()
         db.get_prompt_by_hash.return_value = {"id": 42}
         tracker = PromptTracker(db)
+        self.addCleanup(stop_tracker, self, tracker)
         tracker.set_current_prompt("  Hello World  ")
         db.get_prompt_by_hash.assert_called_once_with(
             generate_prompt_hash("  Hello World  ")

tests/test_image_linking.py

@@ -8,6 +8,7 @@ import unittest.mock as mock
 
 sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
 
+from tests.background_stops import close_handler  # noqa: E402
 from database.operations import PromptDatabase
 from utils.hashing import generate_prompt_hash
 from utils.image_monitor import ImageGenerationHandler
@@ -52,6 +53,7 @@ class LinkingTestCase(unittest.TestCase):
         )
 
     def tearDown(self):
+        close_handler(self, self.handler)  # and the tracker's cleanup thread
         self.db.close_all()  # Windows cannot unlink an open database file
         for path in (
             self.image,

tests/test_image_monitor_singleton.py

@@ -136,13 +136,21 @@ class TestMonitorSingletonLifecycle(unittest.TestCase):
 
 
 class FakeObserver:
-    """Records what the monitor does with the watchdog observer."""
+    """Records what the monitor does with the watchdog observer.
 
-    def __init__(self):
+    ``calls`` lists ``stop`` and ``join`` in order (shared with a test's own
+    recorders when given); ``stay_alive`` keeps it alive after ``join``;
+    ``stop_error`` makes ``stop`` raise.
+    """
+
+    def __init__(self, calls=None, stay_alive=False, stop_error=None):
         self.scheduled = []
         self.started = False
         self.stopped = False
         self.joined = False
+        self.calls = calls if calls is not None else []
+        self.stay_alive = stay_alive
+        self.stop_error = stop_error
 
     def schedule(self, handler, path, recursive=False):
         self.scheduled.append((handler, path, recursive))
@@ -151,12 +159,18 @@ class FakeObserver:
         self.started = True
 
     def stop(self):
+        self.calls.append(("stop",))
+        if self.stop_error is not None:
+            raise self.stop_error
         self.stopped = True
 
     def join(self, timeout=None):
+        self.calls.append(("join", timeout))
         self.joined = True
 
     def is_alive(self):
+        if self.stay_alive:
+            return self.started
         return self.started and not self.stopped
 
 
@@ -171,6 +185,8 @@ class TestStartStop(unittest.TestCase):
         self.missing = os.path.join(self.tmp.name, "does-not-exist")
 
     def tearDown(self):
+        # Before the folder is removed (addCleanup runs after tearDown).
+        self.assertTrue(self.monitor.stop_monitoring())
         im_mod._monitor_instance = None
 
     def test_only_nonexistent_directories_leaves_monitoring_off(self):
@@ -225,9 +241,8 @@ class TestStartStop(unittest.TestCase):
         worker = handler._worker
         self.assertTrue(worker.is_alive())
 
-        self.monitor.stop_monitoring()
+        self.assertTrue(self.monitor.stop_monitoring())
 
-        worker.join(timeout=2)
         self.assertFalse(worker.is_alive())
         self.assertEqual(handler.pending_count(), 0)
         # A restart gets a fresh handler and observer
@@ -303,6 +318,61 @@ class TestStartStop(unittest.TestCase):
         self.assertEqual(self.monitor.monitored_directories, [self.tmp.name])
         self.monitor.stop_monitoring()
 
+    def _start_with(self, fake):
+        with patch.object(im_mod, "Observer", return_value=fake):
+            self.monitor.start_monitoring([self.tmp.name])
+        handler = self.monitor.handler
+        real_close = handler.close
+
+        def close(timeout=im_mod.WORKER_JOIN_SECONDS):
+            fake.calls.append(("close",))
+            return real_close(timeout)
+
+        handler.close = close
+        return handler
+
+    def test_stop_monitoring_joins_observer_and_worker(self):
+        fake = FakeObserver()
+        handler = self._start_with(fake)
+
+        self.assertTrue(self.monitor.stop_monitoring(timeout=3.0))
+
+        self.assertEqual([c[0] for c in fake.calls], ["close", "stop", "join"])
+        join_timeout = fake.calls[2][1]
+        self.assertIsNotNone(join_timeout)
+        self.assertLessEqual(join_timeout, 3.0)
+        self.assertTrue(handler._closed)
+
+    def test_stop_monitoring_reports_an_observer_still_running(self):
+        fake = FakeObserver(stay_alive=True)
+        self._start_with(fake)
+        with self.assertLogs("prompt_manager.image_monitor", "WARNING") as logs:
+            self.assertFalse(self.monitor.stop_monitoring(timeout=0.05))
+        self.assertIn("watchdog observer still running", "\n".join(logs.output))
+        self.assertFalse(self.monitor.is_monitoring)
+        fake.stay_alive = False  # it ends; tearDown's stop waits for it
+
+    def test_a_second_stop_monitoring_waits_for_the_observer(self):
+        fake = FakeObserver(stay_alive=True)
+        self._start_with(fake)
+        with self.assertLogs("prompt_manager.image_monitor", "WARNING"):
+            self.assertFalse(self.monitor.stop_monitoring(timeout=0.05))
+            self.assertFalse(self.monitor.stop_monitoring(timeout=0.05))
+        fake.stay_alive = False
+        self.assertTrue(self.monitor.stop_monitoring())
+        self.assertFalse(self.monitor._unfinished)
+
+    def test_stop_monitoring_reports_an_observer_stop_that_raised(self):
+        fake = FakeObserver(stop_error=OSError("descriptor gone"))
+        self._start_with(fake)
+        with self.assertLogs("prompt_manager.image_monitor", "WARNING") as logs:
+            self.assertFalse(self.monitor.stop_monitoring())
+        self.assertIn("Error while stopping image monitoring", "\n".join(logs.output))
+        fake.stop_error = None  # tearDown's stop calls stop() again
+
+    def test_stop_monitoring_without_an_observer_returns_true(self):
+        self.assertTrue(self.monitor.stop_monitoring())
+
 
 class FakeEvent:
     def __init__(self, path, is_directory=False):
@@ -335,6 +405,10 @@ class TestHandlerScheduling(unittest.TestCase):
             self.processed.append((path, prompt_snapshot))
         )
 
+    def tearDown(self):
+        # Before the folder is removed (addCleanup runs after tearDown).
+        self.assertTrue(self.handler.close())
+
     def _image(self, name, content=b"x" * 16):
         path = os.path.join(self.tmp.name, name)
         with open(path, "wb") as f:
@@ -399,23 +473,127 @@ class TestHandlerScheduling(unittest.TestCase):
         empty = self._image("empty.png", content=b"")
         self.assertFalse(self.handler.wait_until_settled(empty))
 
-    def test_shutdown_stops_the_worker_and_clears_pending(self):
+    def test_shutdown_joins_the_worker_within_the_bound(self):
         self.handler.processing_delay = 10  # keep the first image waiting
         self.handler.schedule(self._image("s1.png"))
         self.handler.schedule(self._image("s2.png"))
         worker = self.handler._worker
         self.assertTrue(worker.is_alive())
 
-        self.handler.shutdown()
+        self.assertTrue(self.handler.shutdown())
 
-        worker.join(timeout=2)
         self.assertFalse(worker.is_alive())
         self.assertEqual(self.handler.pending_count(), 0)
 
+    def test_shutdown_reports_a_worker_still_busy(self):
+        entered, release = threading.Event(), threading.Event()
+
+        def busy(path, prompt_snapshot=None):
+            entered.set()
+            release.wait(30)
+
+        self.handler.process_new_image = busy
+        self.handler.schedule(self._image("busy.png"))
+        worker = self.handler._worker
+        self.addCleanup(worker.join, 30)
+        self.addCleanup(release.set)
+        self.assertTrue(entered.wait(30), "the worker never took the image")
+
+        with self.assertLogs("prompt_manager.image_monitor", "WARNING") as logs:
+            self.assertFalse(self.handler.shutdown(timeout=0.05))
+        self.assertIn("image worker still running", "\n".join(logs.output))
+
+        release.set()
+        worker.join(30)
+        self.assertFalse(worker.is_alive())
+
+    def test_a_second_stop_waits_for_a_worker_still_finishing(self):
+        entered, release = threading.Event(), threading.Event()
+
+        def busy(path, prompt_snapshot=None):
+            entered.set()
+            release.wait(30)
+
+        self.handler.process_new_image = busy
+        self.handler.schedule(self._image("twice.png"))
+        worker = self.handler._worker
+        self.addCleanup(worker.join, 30)
+        self.addCleanup(release.set)
+        self.assertTrue(entered.wait(30), "the worker never took the image")
+
+        with self.assertLogs("prompt_manager.image_monitor", "WARNING"):
+            self.assertFalse(self.handler.close(timeout=0.05))
+            self.assertFalse(self.handler.close(timeout=0.05))
+        release.set()
+        self.assertTrue(self.handler.close(timeout=30))
+        self.assertFalse(worker.is_alive())
+
+    def test_shutdown_on_the_worker_itself_only_signals(self):
+        entered, result = threading.Event(), {}
+
+        def stop_from_inside(path, prompt_snapshot=None):
+            result["value"] = self.handler.shutdown()
+            entered.set()
+
+        self.handler.process_new_image = stop_from_inside
+        self.handler.schedule(self._image("self.png"))
+        worker = self.handler._worker
+        self.assertTrue(entered.wait(30), "the worker never took the image")
+        worker.join(30)
+        self.assertFalse(result["value"])
+        self.assertFalse(worker.is_alive())
+
+    def _live_workers(self):
+        return {
+            t for t in threading.enumerate() if t.name == "PromptManagerImageWorker"
+        }
+
+    def _seal(self, handler, name):
+        handler._await_history(os.path.join(self.tmp.name, name))
+        handler._on_outputs_sealed(
+            "run-1",
+            [{"folder_type": "output", "subfolder": "", "filename": name}],
+        )
+
+    def test_close_refuses_new_work(self):
+        path = self._image("late.png")
+        with patch.object(im_mod, "host_output_directory", return_value=self.tmp.name):
+            self.assertTrue(self.handler.close())
+            before = self._live_workers()
+            self.assertFalse(self.handler.schedule(path))
+            # The image in hand can still ask for a history re-check and its
+            # run can be sealed afterwards, on the history writer's thread.
+            self.handler._await_history(path)
+            self.assertEqual(self.handler.awaiting_history(), [])
+            self._seal(self.handler, "late.png")
+            self.handler._schedule_recheck(path)
+        self.assertIsNone(self.handler._worker)
+        self.assertEqual(list(self.handler._history_due), [])
+        self.assertEqual(self._live_workers() - before, set())
+
+    def test_a_sealed_run_starts_a_worker_on_an_open_handler(self):
+        self._image("open.png")
+        with patch.object(im_mod, "host_output_directory", return_value=self.tmp.name):
+            self._seal(self.handler, "open.png")
+        self.assertIsNotNone(self.handler._worker)
+        self.assertTrue(self.handler.close())
+
+    def test_close_is_idempotent(self):
+        self.handler.schedule(self._image("c1.png"))
+        self.assertTrue(self.handler.close())
+        self.assertTrue(self.handler.close())
+
+    def test_on_created_after_close_takes_no_lock(self):
+        self.assertTrue(self.handler.close())
+        self.handler.on_created(FakeEvent(self._image("after.png")))
+        self.tracker.keyed_snapshot.assert_not_called()
+        self.assertEqual(self.handler.pending_count(), 0)
+
     def test_shutdown_is_idempotent_and_safe_before_start(self):
         fresh = im_mod.ImageGenerationHandler(MagicMock(), self.tracker)
-        fresh.shutdown()
-        fresh.shutdown()
+        self.assertTrue(fresh.shutdown())
+        self.assertTrue(fresh.shutdown())
+        self.assertTrue(fresh.close())
         self.handler.schedule(self._image("s3.png"))
         self.assertTrue(wait_for(lambda: self.handler.pending_count() == 0))
         self.handler.shutdown()
@@ -434,7 +612,7 @@ class TestHandlerScheduling(unittest.TestCase):
         self.assertTrue(wait_for(lambda: self.handler.pending_count() == 0))
         self.assertIn(path, [p for p, _ in self.processed])
         self.assertIsNot(self.handler._worker, old_worker)
-        self.handler.shutdown()
+        self.assertTrue(self.handler.close())
 
     def test_processing_errors_do_not_kill_the_worker(self):
         def explode(path, prompt_snapshot=None):
@@ -449,6 +627,47 @@ class TestHandlerScheduling(unittest.TestCase):
         self.assertEqual(self.processed, [first, second])
 
 
+class TestCloseReleasesHandles(unittest.TestCase):
+    """``close`` joins the worker, so the test folder can be removed (#302)."""
+
+    def test_close_releases_handles(self):
+        from PIL import Image
+
+        from database.operations import PromptDatabase
+        from tests.open_handles import assert_closed
+        from utils.prompt_tracker import PromptTracker
+
+        with tempfile.TemporaryDirectory() as folder:
+            db = PromptDatabase(os.path.join(folder, "prompts.db"))
+            tracker = PromptTracker(db)
+            handler = im_mod.ImageGenerationHandler(db, tracker)
+            handler.processing_delay = 0
+            handler.settle_interval = 0.01
+            done = threading.Event()
+            real_process = handler.process_new_image
+
+            def process(path, prompt_snapshot=None):
+                try:
+                    real_process(path, prompt_snapshot=prompt_snapshot)
+                finally:
+                    done.set()
+
+            handler.process_new_image = process
+            path = os.path.join(folder, "real.png")
+            Image.new("RGB", (8, 8), "red").save(path)
+            try:
+                self.assertTrue(handler.schedule(path))
+                worker = handler._worker
+                self.assertTrue(done.wait(30), "the worker never processed the image")
+            finally:
+                closed, shut = handler.close(), tracker.shutdown()
+                db.model.close_all()
+            self.assertTrue(closed)
+            self.assertTrue(shut)
+            self.assertFalse(worker.is_alive())
+            assert_closed(self, folder)
+
+
 class TestOutputListener(unittest.TestCase):
     """Every settled output is reported to the Images listing (#348, PR 2)."""
 
@@ -456,7 +675,7 @@ class TestOutputListener(unittest.TestCase):
         self.tmp = tempfile.TemporaryDirectory()
         self.addCleanup(self.tmp.cleanup)
         self.handler = im_mod.ImageGenerationHandler(MagicMock(), MagicMock())
-        self.addCleanup(self.handler.shutdown)
+        self.addCleanup(lambda: self.assertTrue(self.handler.close()))
         self.handler.processing_delay = 0
         self.handler.settle_interval = 0.01
         self.handler.process_new_image = lambda path, prompt_snapshot=None: None
@@ -549,7 +768,7 @@ class TestOutputListener(unittest.TestCase):
             self.assertIsNone(monitor.output_listener)
             self.assertIsNone(monitor.handler.output_listener)
         finally:
-            monitor.stop_monitoring()
+            self.assertTrue(monitor.stop_monitoring())
 
 
 class TestModuleIsolation(unittest.TestCase):

tests/test_init_package.py

@@ -4,6 +4,8 @@ The package is loaded the way ComfyUI loads it (``__init__`` with relative
 imports) with ComfyUI's ``server`` and ``folder_paths`` modules stubbed.
 """
 
+import asyncio
+import atexit
 import contextlib
 import importlib.util
 import io
@@ -16,6 +18,7 @@ import threading
 import types
 import unittest
 import unittest.mock as mock
+from pathlib import Path
 
 from aiohttp import web
 
@@ -40,13 +43,20 @@ class FakePromptQueue:
         return self.results.pop(0) if self.results else None
 
 
-def fake_server_module():
-    """A ``server`` module whose PromptServer.instance accepts routes and hooks."""
+def fake_server_module(app=None, queue_task_done=False):
+    """A ``server`` module whose PromptServer.instance accepts routes and hooks.
+
+    ``app`` replaces the default app (one without aiohttp hooks);
+    ``queue_task_done`` gives the queue a ``task_done``, so history registers.
+    """
+    queue = FakePromptQueue()
+    if queue_task_done:
+        queue.task_done = lambda *a, **k: None
     instance = types.SimpleNamespace(
         routes=web.RouteTableDef(),
-        prompt_queue=FakePromptQueue(),
+        prompt_queue=queue,
         handlers=[],
-        app=types.SimpleNamespace(middlewares=[]),
+        app=app if app is not None else types.SimpleNamespace(middlewares=[]),
     )
     instance.add_on_prompt_handler = instance.handlers.append
     module = types.ModuleType("server")
@@ -92,7 +102,7 @@ def _import_package(name, db_path, output_dir, server=None):
             try:
                 spec.loader.exec_module(package)
             except BaseException:
-                cleanup_package(package)
+                cleanup_package(package, require_stopped=False)
                 raise
     finally:
         for key, module in saved.items():
@@ -103,20 +113,34 @@ def _import_package(name, db_path, output_dir, server=None):
     return package, server.PromptServer.instance
 
 
-def cleanup_package(package):
-    """Stop fixture-owned workers before releasing its other resources."""
+def cleanup_package(package, require_stopped=True):
+    """Stop fixture-owned workers before releasing its other resources.
+
+    The package's own teardown runs first (the reload case of #301); with
+    ``require_stopped`` every part it reports must have stopped. The direct
+    stops after it also cover an import that failed before the teardown
+    existed, and a test that made a teardown step fail on purpose.
+    """
+    teardown = getattr(package, "_stop_background_work", None)
+    if callable(teardown):
+        result = teardown()
+        if require_stopped:
+            assert all(result.values()), f"package teardown left work running: {result}"
+    monitor = getattr(package, "_global_image_monitor", None)
+    if monitor is not None:
+        assert monitor.stop_monitoring(), "package image monitor still running"
+    tracker = getattr(package, "_global_prompt_tracker", None)
+    if tracker is not None:
+        assert tracker.shutdown(), "package prompt cleanup thread still running"
     registration = getattr(package, "_history_registration", None)
     if registration is not None:
         assert registration.dispose(3), "package history worker leaked"
-    monitor = getattr(package, "_global_image_monitor", None)
-    if monitor is not None:
-        monitor.stop_monitoring()
     close_every_database()
     sys.modules.pop(package.__name__, None)
 
 
 @contextlib.contextmanager
-def load_package(name, db_path, output_dir, server=None):
+def load_package(name, db_path, output_dir, server=None, require_stopped=True):
     """Dispose package workers before closing its isolated logging handlers."""
     root_logger = logging.getLogger("prompt_manager")
     loggers = [root_logger] + [
@@ -144,7 +168,7 @@ def load_package(name, db_path, output_dir, server=None):
         try:
             yield package, instance
         finally:
-            cleanup_package(package)
+            cleanup_package(package, require_stopped=require_stopped)
     finally:
         current_loggers = [root_logger] + [
             logger
@@ -556,3 +580,241 @@ class TestPackageLoggerIsolation(unittest.TestCase):
         self.assert_logging_restored(
             captured["handlers"], captured["package"]._history_registration
         )
+
+
+PACKAGE_THREAD_NAMES = ("PromptManagerPromptCleanup", "PromptManagerImageWorker")
+
+
+def package_threads():
+    """Live threads this design owns: the tracker, the image worker, and the
+    watchdog observer and emitters (which have no name of their own)."""
+    from watchdog.utils import BaseThread
+
+    return {
+        thread
+        for thread in threading.enumerate()
+        if thread.name in PACKAGE_THREAD_NAMES or isinstance(thread, BaseThread)
+    }
+
+
+def hooked_app():
+    return types.SimpleNamespace(
+        middlewares=[], on_cleanup=[], on_startup=[], on_shutdown=[]
+    )
+
+
+class PackageTeardownCase(unittest.TestCase):
+    """Each test loads its own package (#301): the teardown runs once."""
+
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.tmp.cleanup)
+        self.addCleanup(lambda: self.assertEqual(open_handles_under(self.tmp.name), []))
+        self.output_dir = os.path.join(self.tmp.name, "output")
+        os.makedirs(self.output_dir)
+        self.db_path = os.path.join(self.tmp.name, "prompts.db")
+
+    def load(self, server=None, require_stopped=True, suffix=""):
+        name = f"pm_teardown_{self._testMethodName}{suffix}"
+        stack = contextlib.ExitStack()
+        self.addCleanup(stack.close)
+        stack.enter_context(contextlib.redirect_stdout(io.StringIO()))
+        package, _ = stack.enter_context(
+            load_package(
+                name,
+                self.db_path,
+                self.output_dir,
+                server,
+                require_stopped=require_stopped,
+            )
+        )
+        return package
+
+    def busy_worker(self, package, name="busy.png"):
+        """The image worker alive in its processing delay."""
+        handler = package._global_image_monitor.handler
+        self.assertIsNotNone(handler, "the package's monitor did not start")
+        handler.processing_delay = 3600
+        # Outside the watched folder, so the observer does not schedule it first.
+        path = os.path.join(self.tmp.name, name)
+        with open(path, "wb") as image:
+            image.write(b"x" * 16)
+        self.assertTrue(handler.schedule(path))
+        self.assertTrue(handler._worker.is_alive())
+        return handler._worker
+
+
+class TestPackageTeardown(PackageTeardownCase):
+    def test_teardown_stops_every_part_once(self):
+        package = self.load()
+        monitor = package._global_image_monitor
+        observer = monitor.observer
+        tracker_thread = package._global_prompt_tracker.cleanup_thread
+        worker = self.busy_worker(package)
+
+        with self.assertLogs("prompt_manager.init", "INFO") as logs:
+            result = package._stop_background_work()
+
+        self.assertEqual(result, {"monitor": True, "thumbnails": True, "tracker": True})
+        self.assertEqual(list(result), ["monitor", "thumbnails", "tracker"])
+        self.assertIn("Background work stopped in", "\n".join(logs.output))
+        self.assertIn("monitor=ok thumbnails=ok tracker=ok", "\n".join(logs.output))
+        self.assertFalse(worker.is_alive())
+        self.assertFalse(observer.is_alive())
+        self.assertFalse(tracker_thread.is_alive())
+        result["monitor"] = False  # a copy: the stored result is unchanged
+        with mock.patch.object(monitor, "stop_monitoring") as again:
+            self.assertEqual(
+                package._stop_background_work(),
+                {"monitor": True, "thumbnails": True, "tracker": True},
+            )
+        again.assert_not_called()
+
+    def test_teardown_detaches_the_monitor_callbacks(self):
+        package = self.load()
+        monitor = package._global_image_monitor
+        self.assertIsNotNone(monitor.review_recorder)
+        package._stop_background_work()
+        self.assertIsNone(monitor.review_recorder)
+        self.assertIsNone(monitor.thumbnail_queue)
+        self.assertIsNone(monitor.output_listener)
+
+    def test_an_image_in_hand_queues_no_preview_during_teardown(self):
+        package = self.load()
+        monitor = package._global_image_monitor
+        real_stop = monitor.stop_monitoring
+        image = os.path.join(self.output_dir, "in_hand.png")
+        from PIL import Image
+
+        Image.new("RGB", (8, 8), "red").save(image)
+        seen = {}
+
+        def stop_with_an_image_in_hand(timeout):
+            # The worker read ``thumbnail_queue`` before the detach and now
+            # hands its finished image over (at exit, Python 3.12+ refuses a
+            # new thread here and logs a traceback).
+            seen["queued"] = package.api.queue_capture_thumbnail(image)
+            seen["worker"] = package.api._auto_thumbnailer
+            return real_stop(timeout)
+
+        with (
+            mock.patch.object(
+                package.api,
+                "_get_all_output_dirs",
+                return_value=[Path(os.path.realpath(self.output_dir))],
+            ),
+            mock.patch.object(
+                monitor, "stop_monitoring", side_effect=stop_with_an_image_in_hand
+            ),
+        ):
+            result = package._stop_background_work()
+        self.assertEqual(seen["queued"], "skipped")
+        self.assertIsNone(seen["worker"])
+        self.assertTrue(all(result.values()), result)
+
+    def test_teardown_removes_its_atexit_callback(self):
+        package = self.load()
+        with mock.patch("atexit.unregister", wraps=atexit.unregister) as unregister:
+            package._stop_background_work()
+        unregister.assert_called_once_with(package._stop_background_work)
+
+    def test_on_cleanup_runs_the_teardown(self):
+        app = hooked_app()
+        package = self.load(fake_server_module(app=app))
+        self.assertIs(app.on_cleanup[0], package._stop_background_work_on_cleanup)
+        self.assertIsNone(package._background_work_result)
+
+        async def run_cleanup():
+            for callback in list(app.on_cleanup):
+                await callback(app)
+
+        asyncio.run(run_cleanup())
+        self.assertEqual(
+            package._background_work_result,
+            {"monitor": True, "thumbnails": True, "tracker": True},
+        )
+
+    def test_teardown_runs_before_history_dispose_at_exit(self):
+        with mock.patch("atexit.register", wraps=atexit.register) as register:
+            package = self.load(fake_server_module(queue_task_done=True))
+        registered = [call.args[0] for call in register.call_args_list]
+        registration = package._history_registration
+        self.assertIsNotNone(registration, "history did not register")
+        self.assertIn(registration.dispose, registered)
+        self.assertIn(package._stop_background_work, registered)
+        # atexit runs newest first: the teardown before the history dispose.
+        self.assertGreater(
+            registered.index(package._stop_background_work),
+            registered.index(registration.dispose),
+        )
+
+
+class TestPackageTeardownFailures(PackageTeardownCase):
+    """A failing step never skips the later ones (design D14, R2)."""
+
+    def assert_tracker_stops_after(self, error):
+        package = self.load(require_stopped=False)
+        monitor = package._global_image_monitor
+        tracker_thread = package._global_prompt_tracker.cleanup_thread
+        with (
+            mock.patch.object(monitor, "stop_monitoring", side_effect=error),
+            self.assertLogs("prompt_manager.init", "WARNING") as logs,
+        ):
+            result = package._stop_background_work()
+        self.assertFalse(result["monitor"])
+        self.assertTrue(result["tracker"])
+        self.assertFalse(tracker_thread.is_alive())
+        text = "\n".join(logs.output)
+        self.assertIn("Background work not stopped in time", text)
+        self.assertRegex(text, r"not stopped in time \([0-9.]+ s\): monitor$")
+        self.assertIn("Traceback", text)
+
+    def test_teardown_step_failure_does_not_skip_later_steps(self):
+        self.assert_tracker_stops_after(RuntimeError("stop failed"))
+
+    def test_second_interrupt_does_not_skip_the_tracker(self):
+        self.assert_tracker_stops_after(KeyboardInterrupt())
+
+    def test_teardown_shares_one_deadline(self):
+        package = self.load(require_stopped=False)
+        now = [100.0]
+        seen = []
+
+        def step(timeout, **_):
+            seen.append(timeout)
+            now[0] += 6.0
+            return True
+
+        with (
+            mock.patch.object(package, "_clock", lambda: now[0]),
+            mock.patch.object(
+                package._global_image_monitor, "stop_monitoring", side_effect=step
+            ),
+            mock.patch.object(
+                package.api,
+                "close_auto_thumbnails",
+                side_effect=lambda final=False, timeout=None: step(timeout),
+            ),
+            mock.patch.object(
+                package._global_prompt_tracker, "shutdown", side_effect=step
+            ),
+        ):
+            package._stop_background_work()
+        self.assertEqual(seen, [5.0, 4.0, 0.0])
+
+
+class TestRepeatedLoads(PackageTeardownCase):
+    def test_repeated_loads_leave_no_threads(self):
+        for round_ in range(3):
+            before = package_threads()
+            with contextlib.redirect_stdout(io.StringIO()):
+                with load_package(
+                    f"pm_teardown_repeat_{round_}",
+                    self.db_path,
+                    self.output_dir,
+                ) as (package, _):
+                    self.busy_worker(package, f"busy_{round_}.png")
+                    started = package_threads() - before
+                    self.assertTrue(started, "the package started no thread")
+            left = {t for t in package_threads() - before if t.is_alive()}
+            self.assertEqual(left, set(), f"round {round_}")

tests/test_library_move_review.py

@@ -29,6 +29,7 @@ sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
 from database import library_move as lm  # noqa: E402
 from database import models  # noqa: E402
 from database.operations import PromptDatabase  # noqa: E402
+from tests.background_stops import close_handler  # noqa: E402
 from tests.open_handles import assert_closed  # noqa: E402
 from utils import library_location as ll  # noqa: E402
 from utils import usage_tracking  # noqa: E402
@@ -204,6 +205,7 @@ class TestBackgroundWritesSurviveTheSwitch(_MoveCase):
         image = os.path.join(self.comfy, "output", "late.png")
         writer_db = PromptDatabase(self.old_db)
         handler = ImageGenerationHandler(writer_db, mock.MagicMock())
+        self.addCleanup(close_handler, self, handler)
         started = threading.Event()
         probe = _CrossThreadCloseProbe()
         outcome = {}
@@ -256,6 +258,7 @@ class TestBackgroundWritesSurviveTheSwitch(_MoveCase):
 
         db = PromptDatabase(self.old_db)
         handler = ImageGenerationHandler(db, mock.MagicMock())
+        self.addCleanup(close_handler, self, handler)
         image = os.path.join(self.comfy, "output", "orphan.png")
         with mock.patch.object(handler, "logger") as log:
             handler.link_image_to_prompt(image, {"id": 99999}, {})

tests/test_nonfinite_metadata.py

@@ -24,6 +24,7 @@ from unittest.mock import patch
 from PIL import Image, PngImagePlugin
 
 from database.operations import PromptDatabase
+from tests.background_stops import close_handler
 from tests.history_waits import drain_writer
 from tests.test_api_admin import AdminAPITestCase
 from tests.test_api_images import ImageAPITestCase
@@ -224,6 +225,7 @@ class TestMonitorIngestsNonFiniteMetadata(StrictRows, unittest.TestCase):
         self.handler.logger.disabled = True
 
     def tearDown(self):
+        close_handler(self, self.handler)
         self.db.close_all()
         assert_closed(self, self.temp.name)
         self.temp.cleanup()

tests/test_package_exit.py

@@ -0,0 +1,142 @@
+"""The package teardown runs at a normal interpreter exit (#301, #302).
+
+ComfyUI catches Ctrl+C in ``main.py`` and returns, so its exit is a normal
+return: ``atexit`` callbacks then run, newest first. A child process imports
+the package with ComfyUI stubbed, leaves the image worker alive in its
+processing delay and returns. Its own ``atexit`` reporter, registered before
+the package, runs after the teardown and prints the result as one JSON line.
+"""
+
+import json
+import os
+import subprocess
+import sys
+import tempfile
+import textwrap
+import unittest
+
+ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+# The parent never waits longer than this for the child (the hang watchdog
+# allows 120 s per test).
+CHILD_CEILING_SECONDS = 60
+REPORT_PREFIX = "PM_EXIT_REPORT "
+
+CHILD = textwrap.dedent("""
+    import atexit
+    import contextlib
+    import io
+    import json
+    import os
+    import sys
+    import threading
+
+    root, tmp = sys.argv[1], sys.argv[2]
+    sys.path.insert(0, root)
+    state = {}
+
+    def report():
+        from watchdog.utils import BaseThread
+
+        package = state.get("package")
+        result = getattr(package, "_background_work_result", None)
+        # The image worker, the tracker, and watchdog's observer and emitters.
+        alive = sorted(
+            t.name
+            for t in threading.enumerate()
+            if t.name in ("PromptManagerImageWorker", "PromptManagerPromptCleanup")
+            or isinstance(t, BaseThread)
+        )
+        print("PM_EXIT_REPORT " + json.dumps({"result": result, "alive": alive}))
+        sys.stdout.flush()
+
+    # First registered, so it runs last: after the package's teardown and
+    # after the history dispose.
+    atexit.register(report)
+
+    from tests.test_init_package import _import_package, fake_server_module
+
+    output = os.path.join(tmp, "output")
+    with contextlib.redirect_stdout(io.StringIO()):
+        package, _ = _import_package(
+            "pm_exit_child",
+            os.path.join(tmp, "prompts.db"),
+            output,
+            fake_server_module(queue_task_done=True),
+        )
+    state["package"] = package
+    monitor = package._global_image_monitor
+    handler = monitor.handler if monitor is not None else None
+    if handler is None:
+        sys.exit(3)
+    handler.processing_delay = 3600
+    path = os.path.join(tmp, "busy.png")  # outside the watched folder
+    with open(path, "wb") as image:
+        image.write(b"x" * 16)
+    if not handler.schedule(path) or not handler._worker.is_alive():
+        sys.exit(4)
+    # Return normally with the worker alive, as ComfyUI does after Ctrl+C.
+    """)
+
+
+class TestPackageExit(unittest.TestCase):
+    def test_teardown_runs_at_a_normal_exit_with_the_worker_alive(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            os.makedirs(os.path.join(tmp, "output"))
+            script = os.path.join(tmp, "child.py")
+            with open(script, "w", encoding="utf-8") as handle:
+                handle.write(CHILD)
+            env = dict(os.environ)
+            env.pop("PROMPT_MANAGER_CONFIG_PATH", None)
+            # The child's library and log file live in its own folder, never
+            # in the checkout's logs/ (which this runner may hold open) or a
+            # data folder the caller set.
+            env["PROMPT_MANAGER_DATA_DIR"] = tmp
+            env["PYTHONIOENCODING"] = "utf-8"
+            try:
+                completed = subprocess.run(
+                    [sys.executable, script, ROOT, tmp],
+                    cwd=ROOT,
+                    env=env,
+                    capture_output=True,
+                    timeout=CHILD_CEILING_SECONDS,
+                    encoding="utf-8",
+                    errors="replace",
+                )
+            except subprocess.TimeoutExpired as hung:
+                self.fail(
+                    f"the child did not exit within {CHILD_CEILING_SECONDS} s\n"
+                    f"stdout:\n{hung.stdout}\nstderr:\n{hung.stderr}"
+                )
+            # Leaving the block removes the folder: on Windows that fails
+            # if the child left a handle open, which it cannot once it exited,
+            # so the check here is that the child exited at all.
+        detail = f"stdout:\n{completed.stdout}\nstderr:\n{completed.stderr}"
+        self.assertEqual(completed.returncode, 0, detail)
+        reports = [
+            line[len(REPORT_PREFIX) :]
+            for line in completed.stdout.splitlines()
+            if line.startswith(REPORT_PREFIX)
+        ]
+        self.assertEqual(len(reports), 1, detail)
+        report = json.loads(reports[0])
+        # "thumbnails" is True here because no preview worker started; its
+        # close result is tested in test_thumbnail_auto and test_init_package.
+        self.assertEqual(
+            report["result"],
+            {"monitor": True, "thumbnails": True, "tracker": True},
+            detail,
+        )
+        self.assertEqual(report["alive"], [], detail)
+        # The teardown's own warnings. Import-time warnings (the API's
+        # output-folder search finds no ComfyUI here) are not about exit.
+        for text in (
+            "Traceback",
+            "Background work not stopped in time",
+            "still running after",
+            "Error while stopping",
+        ):
+            self.assertNotIn(text, completed.stderr, detail)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_prompt_attribution.py

@@ -23,6 +23,7 @@ try:
 except ImportError:  # discovered with tests/ as the top-level directory
     from open_handles import assert_closed
 
+from tests.background_stops import close_handler  # noqa: E402
 from database.operations import PromptDatabase
 from prompt_manager import PromptManager
 from prompt_manager_pair import PromptManagerPair
@@ -278,6 +279,7 @@ class MonitorCase(DatabaseCase):
         fd, self.image = tempfile.mkstemp(suffix=".png", dir=self.tmp.name)
         os.close(fd)
         self.handler = ImageGenerationHandler(self.db, self.tracker)
+        self.addCleanup(close_handler, self, self.handler)
         self.linked = []
         self.handler.link_image_to_prompt = lambda path, prompt, meta: (
             self.linked.append(prompt["id"])

tests/test_prompt_tracker.py

@@ -5,10 +5,12 @@ Tests PromptTracker, PromptExecutionContext, and the singleton
 get_prompt_tracker. Uses a mock db_manager to avoid database deps.
 """
 
+import logging
 import os
 import sys
 import time
 import threading
+import types
 import unittest
 from unittest.mock import MagicMock, patch
 
@@ -312,5 +314,93 @@ class TestPromptQueue(unittest.TestCase):
         self.assertEqual(popped_ids, ids)
 
 
+def _real_tracker():
+    """A PromptTracker built by its own ``__init__``, so its cleanup thread runs."""
+    mock_db = MagicMock()
+    mock_db.get_prompt_by_hash.return_value = None
+    return PromptTracker(mock_db)
+
+
+class TestTrackerShutdown(unittest.TestCase):
+    """The cleanup thread stops on a signal and is joined within a bound (#301)."""
+
+    def test_shutdown_stops_the_cleanup_thread_within_the_bound(self):
+        tracker = _real_tracker()
+        thread = tracker.cleanup_thread
+        self.assertTrue(thread.is_alive())
+        self.assertEqual(thread.name, "PromptManagerPromptCleanup")
+
+        self.assertTrue(tracker.shutdown())
+        self.assertFalse(thread.is_alive())
+
+    def test_shutdown_is_idempotent(self):
+        tracker = _real_tracker()
+        self.assertTrue(tracker.shutdown())
+        self.assertTrue(tracker.shutdown())
+
+    def test_cleanup_error_wait_is_interruptible(self):
+        import utils.prompt_tracker as tracker_mod
+
+        real_time = time.time
+        calls = {"n": 0}
+
+        def failing_once():
+            calls["n"] += 1
+            if calls["n"] == 1:
+                raise RuntimeError("clock unavailable")
+            return real_time()
+
+        stub = types.SimpleNamespace(
+            time=failing_once, sleep=time.sleep, monotonic=time.monotonic
+        )
+        logged = threading.Event()
+        logger = logging.getLogger("prompt_manager.prompt_tracker")
+        real_error = logger.error
+
+        def error(*args, **kwargs):
+            real_error(*args, **kwargs)
+            logged.set()
+
+        with (
+            patch.object(tracker_mod, "time", stub),
+            patch.object(logger, "error", side_effect=error),
+        ):
+            tracker = _real_tracker()
+            self.addCleanup(tracker.shutdown)
+            self.assertTrue(logged.wait(30), "the cleanup error was never logged")
+            # The thread is now in its 60 s error wait.
+            self.assertTrue(tracker.shutdown())
+        self.assertFalse(tracker.cleanup_thread.is_alive())
+
+    def test_tracker_works_after_shutdown(self):
+        tracker = _real_tracker()
+        self.assertTrue(tracker.shutdown())
+        tracker.set_current_prompt("still tracked", {"prompt_id": 9})
+        self.assertEqual(tracker.pop_next_prompt()["id"], 9)
+
+    def test_shutdown_of_a_tracker_built_without_init(self):
+        self.assertTrue(_make_tracker().shutdown())
+
+    def test_shutdown_on_the_cleanup_thread_only_signals(self):
+        tracker = _real_tracker()
+        self.addCleanup(lambda: self.assertTrue(tracker.shutdown()))
+        result = {}
+        with patch.object(tracker, "cleanup_thread", threading.current_thread()):
+            result["value"] = tracker.shutdown()
+        self.assertFalse(result["value"])
+        self.assertTrue(tracker._stop.is_set())
+
+    def test_shutdown_reports_a_thread_still_running(self):
+        tracker = _real_tracker()
+        self.assertTrue(tracker.shutdown())
+        busy = MagicMock()
+        busy.is_alive.return_value = True
+        tracker.cleanup_thread = busy
+        with self.assertLogs("prompt_manager.prompt_tracker", "WARNING") as logs:
+            self.assertFalse(tracker.shutdown(timeout=0.05))
+        busy.join.assert_called_once_with(0.05)
+        self.assertIn("prompt cleanup thread still running", "\n".join(logs.output))
+
+
 if __name__ == "__main__":
     unittest.main()

tests/test_review.py

@@ -27,6 +27,7 @@ except ImportError:  # discovered with tests/ as the top-level directory
 
 from PIL import Image, PngImagePlugin
 
+from tests.background_stops import close_handler, stop_monitor
 from database.operations import PromptDatabase
 from database.review_schema import canonical_evidence, evidence_digest
 from py.api import PromptManagerAPI
@@ -790,6 +791,7 @@ class TestReviewEdges(ReviewTestCase):
 class TestMonitorRecording(ReviewTestCase):
     def _handler(self):
         handler = ImageGenerationHandler(self.api.db, PromptTracker(self.api.db))
+        self.addCleanup(close_handler, self, handler)
         return handler
 
     def _linked_prompt(self, path):
@@ -953,7 +955,7 @@ class TestMonitorRecording(ReviewTestCase):
             monitor.set_review_recorder(None)
             self.assertIsNone(monitor.handler.review_recorder)
         finally:
-            monitor.stop_monitoring()
+            stop_monitor(self, monitor)
 
 
 if __name__ == "__main__":
@@ -1092,6 +1094,7 @@ class TestCaptureStoreShutdown(ReviewTestCase):
                 self._scan_record(path)
             # The monitor's pass records nothing, so it needs no store.
             handler = ImageGenerationHandler(self.api.db, PromptTracker(self.api.db))
+            self.addCleanup(close_handler, self, handler)
             handler.review_recorder = self.api.record_review_file
             handler._record_review(str(path))
             self.assertEqual(self.api.record_review_file(str(path)), "skipped")

tests/test_review_image_links_tool.py

@@ -17,6 +17,7 @@ except ImportError:  # discovered with tests/ as the top-level directory
     from open_handles import assert_closed
 
 from database.operations import PromptDatabase
+from tests.background_stops import close_handler
 from tests.test_api_images import ImageAPITestCase
 from tests.test_prompt_attribution import issue_graph, pair_evidence
 from tools import review_image_links as tool
@@ -158,8 +159,8 @@ class ReviewToolTests(ToolFixture):
 
     def test_conflicting_sources_are_left_to_the_maintainer(self):
         db = PromptDatabase(self.db_path)
+        handler = tool.ImageGenerationHandler(db, None)
         try:
-            handler = tool.ImageGenerationHandler(db, None)
             row = {
                 "id": 1,
                 "prompt_id": self.coffee,
@@ -170,6 +171,7 @@ class ReviewToolTests(ToolFixture):
             ):
                 verdict = tool.judge(db, handler, row, self.output)
         finally:
+            close_handler(self, handler)
             db.close_all()
         self.assertEqual(verdict["verdict"], "conflict")
 

tests/test_review_inspection.py

@@ -28,6 +28,7 @@ except ImportError:  # discovered with tests/ as the top-level directory
     from test_api_review import ReviewRoutesTestCase
     from test_review import unlinkable_graph, write_png
 
+from tests.background_stops import close_handler
 from utils import inspection_semantics, review_capture
 from utils.hashing import generate_prompt_hash
 from utils.image_monitor import ImageGenerationHandler
@@ -437,6 +438,7 @@ class TestApplyInspectionCandidate(InspectionTestCase):
 class TestMonitorNeverQueues(InspectionTestCase):
     def _handler(self):
         handler = ImageGenerationHandler(self.api.db, PromptTracker(self.api.db))
+        self.addCleanup(close_handler, self, handler)
         handler.review_recorder = self.api.record_review_file
         return handler
 
@@ -760,6 +762,7 @@ class TestRecordingEdges(InspectionTestCase):
             text="latest", prompt_hash=generate_prompt_hash("latest")
         )
         handler = ImageGenerationHandler(self.api.db, PromptTracker(self.api.db))
+        self.addCleanup(close_handler, self, handler)
         handler.link_image_to_prompt(
             str(path),
             {"id": prompt_id, "text": "latest", "fallback": True},

tests/test_review_outcome.py

@@ -38,6 +38,7 @@ except ImportError:  # discovered with tests/ as the top-level directory
         write_png,
     )
 
+from tests.background_stops import close_handler
 from utils.hashing import generate_prompt_hash
 from utils.image_monitor import ImageGenerationHandler
 from utils.prompt_tracker import PromptTracker
@@ -299,6 +300,7 @@ class TestQueueSelfHeals(OutcomeTestCase):
 class TestMonitorRule(OutcomeTestCase):
     def _handler(self):
         handler = ImageGenerationHandler(self.api.db, PromptTracker(self.api.db))
+        self.addCleanup(close_handler, self, handler)
         handler.review_recorder = self.api.record_review_file
         return handler
 

tests/test_run_attribution.py

@@ -11,6 +11,7 @@ from unittest import mock
 
 from database.run_attribution import stored_run_prompt
 from tests import test_execution_history_database as history_tests
+from tests.background_stops import close_handler
 from tests.test_api_admin import AdminAPITestCase
 from tests.test_execution_context_database import snapshot
 from tests.test_prompt_attribution import issue_graph, llm_text_graph, make_tracker
@@ -151,6 +152,7 @@ class TestStoredRunPrompt(RecordedRuns, unittest.TestCase):
         prompt_id = self.save(LLM_TEXT)
         self.record_run(self.db, self.output, [("sub", "img_00001_.png")], [LLM_TEXT])
         handler = ImageGenerationHandler(self.db, make_tracker(self.db))
+        self.addCleanup(close_handler, self, handler)
         linked = []
         handler.link_image_to_prompt = lambda path, prompt, meta: linked.append(
             prompt["id"]
@@ -332,7 +334,7 @@ class TestMonitorHistoryRecheck(RecordedRuns, unittest.TestCase):
         self.handler = ImageGenerationHandler(self.db, make_tracker(self.db))
         self.handler.processing_delay = 0
         self.handler.settle_interval = 0.01
-        self.addCleanup(self.handler.shutdown)
+        self.addCleanup(close_handler, self, self.handler)
         self.linked = []
         self.handler.link_image_to_prompt = lambda path, prompt, meta: (
             self.linked.append((os.path.basename(path), prompt["id"]))
@@ -650,7 +652,7 @@ class TestLinkedOutputRoot(RecordedRuns, unittest.TestCase):
         handler = ImageGenerationHandler(db, make_tracker(db))
         handler.processing_delay = 0
         handler.settle_interval = 0.01
-        self.addCleanup(handler.shutdown)
+        self.addCleanup(close_handler, self, handler)
         linked = []
         handler.link_image_to_prompt = lambda path, prompt, meta: linked.append(
             prompt["id"]

tests/test_settings_sources.py

@@ -1037,6 +1037,23 @@ class TestAutotagDefaultsSettings(SourcesTestCase):
         settings = (await self._get())["settings"]
         self.assertEqual({key: settings[key] for key in self.SAVED}, self.SAVED)
 
+    async def test_threshold_near_the_grid_is_saved_on_it(self):
+        """#369: a computed value a hair off the 0.05 grid is stored on it, so
+        every page's step check accepts what /settings and the file return."""
+        revision = (await self._get())["revision"]
+        status, data = await self._post(
+            "/prompt_manager/settings",
+            {"expected_revision": revision, "autotag_general_threshold": 0.1 + 0.2},
+        )
+        self.assertEqual(status, 200, data)
+        self.assertEqual(
+            repr(self._read_config()["autotag"]["defaults"]["general_threshold"]),
+            "0.3",
+        )
+        self.assertEqual(
+            repr((await self._get())["settings"]["autotag_general_threshold"]), "0.3"
+        )
+
     async def test_stale_revision_changes_no_default(self):
         revision = (await self._get())["revision"]
         self._write_config({"custom": {"edited": "elsewhere"}})
@@ -1055,6 +1072,16 @@ class TestAutotagDefaultsSettings(SourcesTestCase):
                 1.5,
                 "General-tag confidence must be between 0 and 1.",
             ),
+            (
+                "autotag_general_threshold",
+                0.37,
+                "General-tag confidence must be a multiple of 0.05.",
+            ),
+            (
+                "autotag_character_threshold",
+                0.33,
+                "Character-tag confidence must be a multiple of 0.05.",
+            ),
             (
                 "autotag_character_threshold",
                 "0.8",

tests/test_thumbnail_auto.py

@@ -32,6 +32,7 @@ from PIL import Image  # noqa: E402
 
 import py.api.images as images_module  # noqa: E402
 import py.api.thumbnail_auto as auto  # noqa: E402
+from tests.background_stops import close_handler, stop_monitor  # noqa: E402
 from py.api.jobs import THUMBNAILS_JOB_KIND  # noqa: E402
 
 SERVE = "/prompt_manager/images/serve/"
@@ -737,11 +738,27 @@ class TestShutdown(AutoThumbnailCase):
             patch.object(auto, "CLOSE_WAIT_SECONDS", 0),
             self.assertLogs("prompt_manager.api", "WARNING") as logs,
         ):
-            self.api.close_auto_thumbnails()
+            self.assertFalse(self.api.close_auto_thumbnails())
         process.kill.assert_called_once()
         self.assertIn("still running", "\n".join(logs.output))
         release.set()
 
+    async def test_close_reports_an_ended_worker(self):
+        path = make_png(self.output_dir / "a.png", size=(64, 64))
+        self.api.queue_capture_thumbnail(str(path))
+        worker = self.api._auto_thumbnails()._worker
+        await self._idle()
+        loop = asyncio.get_running_loop()
+        closed = await loop.run_in_executor(
+            None, lambda: self.api.close_auto_thumbnails(timeout=WAIT)
+        )
+        self.assertTrue(closed)
+        self.assertFalse(worker.is_alive())
+
+    def test_close_without_a_thumbnailer_returns_true(self):
+        self.assertIsNone(self.api._auto_thumbnailer)
+        self.assertTrue(self.api.close_auto_thumbnails())
+
 
 class TestHistorySharesTheAdmission(AutoThumbnailCase):
     async def _shared_flight(self, name, is_video, first):
@@ -980,6 +997,7 @@ class TestMonitorQueuesThumbnails(AutoThumbnailCase):
         from utils.image_monitor import ImageGenerationHandler
 
         handler = ImageGenerationHandler(self.api.db, unittest.mock.MagicMock())
+        self.addCleanup(close_handler, self, handler)
         handler.processing_delay = 0
         handler.settle_interval = 0
         return handler
@@ -1022,7 +1040,7 @@ class TestMonitorQueuesThumbnails(AutoThumbnailCase):
             monitor.set_thumbnail_queue(None)
             self.assertIsNone(monitor.handler.thumbnail_queue)
         finally:
-            monitor.stop_monitoring()
+            stop_monitor(self, monitor)
 
 
 class TestScanHandsOffThumbnails(AutoThumbnailCase):

tests/test_usage_tracking.py

@@ -16,6 +16,7 @@ import uuid
 
 sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
 
+from tests.background_stops import stop_tracker  # noqa: E402
 from database.operations import PromptDatabase
 from prompt_manager_base import PromptManagerBase
 from utils.hashing import generate_prompt_hash
@@ -596,6 +597,7 @@ class TestNodeRoles(DbTestCase):
 class TestTrackerQueueFlag(unittest.TestCase):
     def setUp(self):
         self.tracker = PromptTracker(mock.Mock())
+        self.addCleanup(stop_tracker, self, self.tracker)
 
     def test_push_to_queue_false_keeps_the_batch_queue_empty(self):
         self.tracker.set_current_prompt("a cat", {"prompt_id": 1}, push_to_queue=False)

utils/image_monitor.py

@@ -70,6 +70,11 @@ MAX_AWAITING_HISTORY = 1000
 AWAIT_HISTORY_SECONDS = 3600
 # ``schedule`` snapshot meaning "history-only re-check" (not a new image)
 HISTORY_ONLY = object()
+# How long a stop waits for the image worker, and for the watchdog observer
+# after its stop (#302). Every wait before the image in hand is interrupted
+# at once, so a bound is reached only while one image is still being linked.
+WORKER_JOIN_SECONDS = 5.0
+OBSERVER_JOIN_SECONDS = 5.0
 
 
 class ImageGenerationHandler(FileSystemEventHandler):
@@ -143,6 +148,11 @@ class ImageGenerationHandler(FileSystemEventHandler):
         self._history_due: "OrderedDict[str, None]" = OrderedDict()
         self._awaiting_lock = threading.Lock()
         self._sealed_subscribed = False
+        # Set by ``close``: no new work, no new worker, ever (#302).
+        self._closed = False
+        # Workers a stop signalled that had not ended yet: every later stop
+        # waits for them again, so it never reports True too early.
+        self._finishing: List[threading.Thread] = []
 
     def on_created(self, event):
         """Handle filesystem creation events.
@@ -157,6 +167,10 @@ class ImageGenerationHandler(FileSystemEventHandler):
         Args:
             event: FileSystemEvent object containing event details
         """
+        # Closed: return before any log or tracker lock, so watchdog's
+        # ``observer.stop()`` never waits on this dispatch.
+        if self._closed:
+            return
         if event.is_directory or not self.is_image_file(event.src_path):
             return
         self.logger.info(f"New image detected: {event.src_path}")
@@ -180,9 +194,12 @@ class ImageGenerationHandler(FileSystemEventHandler):
         and once its size has stopped changing.
 
         Returns:
-            False when the path is already pending or the pending set is full
+            False when the path is already pending, the pending set is full or
+            the handler is closed
         """
         with self._pending_lock:
+            if self._closed:
+                return False
             if image_path in self._pending:
                 self.logger.debug(f"Already pending, not rescheduled: {image_path}")
                 return False
@@ -203,12 +220,19 @@ class ImageGenerationHandler(FileSystemEventHandler):
         with self._pending_lock:
             return len(self._pending)
 
-    def shutdown(self) -> None:
-        """Stop the worker thread and drop every image still waiting.
+    def shutdown(self, timeout: float = WORKER_JOIN_SECONDS) -> bool:
+        """Stop the worker thread, drop every image still waiting, and wait
+        for the worker within ``timeout`` seconds.
 
-        The worker wakes from any delay or settle wait, finishes nothing more
-        and exits; the next ``schedule`` starts a fresh worker with its own
-        queue. Safe to call repeatedly and before anything was scheduled.
+        The worker wakes from any delay or settle wait, finishes at most the
+        image in hand and exits; the next ``schedule`` starts a fresh worker
+        with its own queue (use ``close`` for a final stop). Safe to call
+        repeatedly and before anything was scheduled.
+
+        Returns:
+            True when no worker is left running. False when the worker is
+            still finishing its image after ``timeout`` (a warning is
+            logged), or when called on the worker itself (it only signals).
         """
         with self._awaiting_lock:
             self._awaiting.clear()
@@ -218,9 +242,40 @@ class ImageGenerationHandler(FileSystemEventHandler):
             worker, self._worker = self._worker, None
             work, self._work = self._work, queue.Queue()
             stop, self._stop = self._stop, threading.Event()
+            if worker is not None:
+                self._finishing.append(worker)
+            finishing = [w for w in self._finishing if w.is_alive()]
+            self._finishing = finishing
         stop.set()
         if worker is not None and worker.is_alive():
             work.put(None)  # sentinel: wake a worker idle on an empty queue
+        if threading.current_thread() in finishing:
+            self.logger.debug("shutdown called on the image worker itself")
+            return False
+        deadline = time.monotonic() + timeout
+        for thread in finishing:
+            thread.join(max(0.0, deadline - time.monotonic()))
+        with self._pending_lock:
+            self._finishing = [w for w in self._finishing if w.is_alive()]
+            still_running = bool(self._finishing)
+        if still_running:
+            self.logger.warning(
+                f"image worker still running after {timeout} s; "
+                "it stops after the image in hand"
+            )
+            return False
+        return True
+
+    def close(self, timeout: float = WORKER_JOIN_SECONDS) -> bool:
+        """Final stop: ``shutdown`` and refuse every later ``schedule``.
+
+        After ``close`` no worker starts again, not even from a sealed run's
+        history re-check on the history writer's thread. Returns what
+        ``shutdown`` returns.
+        """
+        with self._pending_lock:
+            self._closed = True
+        return self.shutdown(timeout)
 
     def _ensure_worker(self) -> None:
         """Start the single daemon worker thread if it isn't running.
@@ -521,7 +576,12 @@ class ImageGenerationHandler(FileSystemEventHandler):
             self._awaiting.popitem(last=False)
 
     def _await_history(self, image_path: str) -> None:
-        """Re-check ``image_path`` once, history only, when its run is sealed."""
+        """Re-check ``image_path`` once, history only, when its run is sealed.
+
+        A closed handler re-checks nothing.
+        """
+        if self._closed:
+            return
         if not self._sealed_subscribed:
             subscribe_outputs_sealed(self._on_outputs_sealed)
             self._sealed_subscribed = True
@@ -570,10 +630,19 @@ class ImageGenerationHandler(FileSystemEventHandler):
             self._schedule_recheck(image_path)
 
     def _schedule_recheck(self, image_path: str) -> None:
-        """Schedule the history-only re-check, or keep it due for later."""
+        """Schedule the history-only re-check, or keep it due for later.
+
+        A closed handler drops it.
+        """
+        if self._closed:
+            return
         if self.schedule(image_path, HISTORY_ONLY):
             return
         with self._awaiting_lock:
+            # ``close`` sets the flag before ``shutdown`` clears this set, so
+            # a close that raced the refusal above is seen here.
+            if self._closed:
+                return
             self._history_due.pop(image_path, None)
             self._history_due[image_path] = None
             while len(self._history_due) > MAX_AWAITING_HISTORY:
@@ -872,6 +941,9 @@ class ImageMonitor:
         self.observer = None
         self.handler = None
         self.monitored_directories = []
+        # (observer, handler) pairs a stop left running: the next stop waits
+        # for them again instead of reporting True.
+        self._unfinished: List[Tuple[Any, Any]] = []
         self.logger = get_logger("prompt_manager.image_monitor")
         self.review_recorder: Optional[Callable[[str], str]] = None
         self.thumbnail_queue: Optional[Callable[[str], str]] = None
@@ -999,29 +1071,69 @@ class ImageMonitor:
         """Whether a filesystem observer is running."""
         return self.observer is not None
 
-    def stop_monitoring(self):
-        """Stop the image monitoring system.
+    def stop_monitoring(self, timeout: float = OBSERVER_JOIN_SECONDS) -> bool:
+        """Stop the image monitoring system within one ``timeout`` deadline.
 
-        Cleanly shuts down the filesystem watcher and clears all monitoring state.
-        This method should be called before program exit to ensure proper cleanup.
-        Safe to call when monitoring never started.
+        Closes the handler first (its worker is joined and its dispatch
+        returns at once), then stops the watchdog observer and joins it.
+        watchdog's ``observer.stop()`` joins its emitters without a limit;
+        it counts against the deadline but is not bounded here. A part an
+        earlier call left running is waited for again. Safe to call when
+        monitoring never started and more than once.
+
+        Returns:
+            True when every worker and observer it stopped has ended.
         """
+        deadline = time.monotonic() + timeout
         observer, self.observer = self.observer, None
         handler, self.handler = self.handler, None
         self.monitored_directories = []
+        pairs, self._unfinished = self._unfinished, []
+        if observer is not None:
+            pairs.append((observer, handler))
+        if not pairs:
+            return True
+
+        def remaining() -> float:
+            return max(0.0, deadline - time.monotonic())
+
+        for pair_observer, pair_handler in pairs:
+            handler_ok = self._close_handler(pair_handler, remaining())
+            observer_ok = self._stop_observer(pair_observer, remaining())
+            if not observer_ok:
+                self.logger.warning(
+                    f"watchdog observer still running after {timeout} s"
+                )
+            if not (handler_ok and observer_ok):
+                self._unfinished.append(
+                    (
+                        None if observer_ok else pair_observer,
+                        None if handler_ok else pair_handler,
+                    )
+                )
+        self.logger.debug("Image monitoring stopped")
+        return not self._unfinished
+
+    def _close_handler(self, handler, timeout: float) -> bool:
+        if handler is None:
+            return True
+        try:
+            return handler.close(timeout)
+        except Exception as e:
+            self.logger.warning(f"Error while stopping the image worker: {e}")
+            return False
+
+    def _stop_observer(self, observer, timeout: float) -> bool:
+        """``observer.stop()`` (safe to repeat), then a bounded join."""
         if observer is None:
-            return
+            return True
         try:
             observer.stop()
-            observer.join()
+            observer.join(timeout)
         except Exception as e:
             self.logger.warning(f"Error while stopping image monitoring: {e}")
-        if handler is not None:
-            try:
-                handler.shutdown()
-            except Exception as e:
-                self.logger.warning(f"Error while stopping the image worker: {e}")
-        self.logger.debug("Image monitoring stopped")
+            return False
+        return not observer.is_alive()
 
     def detect_comfyui_output_dirs(self) -> list:
         """Auto-detect ComfyUI output directories.

utils/prompt_tracker.py

@@ -44,6 +44,9 @@ except ImportError:
     from utils.hashing import generate_prompt_hash
     from utils.logging_config import get_logger
 
+# How long ``PromptTracker.shutdown`` waits for the cleanup thread (#301).
+TRACKER_JOIN_SECONDS = 2.0
+
 
 class PromptTracker:
     """Thread-safe tracking of current prompt executions.
@@ -92,9 +95,12 @@ class PromptTracker:
             self.cleanup_interval = 300  # 5 minutes
             self.prompt_timeout = 600  # 10 minutes
 
-        # Start cleanup thread
+        # Start cleanup thread; ``shutdown`` sets ``_stop`` to end it.
+        self._stop = threading.Event()
         self.cleanup_thread = threading.Thread(
-            target=self._cleanup_expired_prompts, daemon=True
+            target=self._cleanup_expired_prompts,
+            name="PromptManagerPromptCleanup",
+            daemon=True,
         )
         self.cleanup_thread.start()
         self.logger.debug("PromptTracker initialization completed")
@@ -464,9 +470,10 @@ class PromptTracker:
 
         Runs continuously as a daemon thread, periodically removing expired
         prompt contexts from the global tracking dictionary. This prevents
-        memory leaks from accumulating old prompt data.
+        memory leaks from accumulating old prompt data. Every wait ends at
+        once when ``shutdown`` sets the stop event.
         """
-        while True:
+        while not self._stop.is_set():
             try:
                 current_time = time.time()
                 expired_ids = []
@@ -497,11 +504,37 @@ class PromptTracker:
                     if pruned:
                         self.logger.debug(f"Pruned {pruned} expired queue entries")
 
-                time.sleep(self.cleanup_interval)
+                if self._stop.wait(self.cleanup_interval):
+                    break
 
             except Exception as e:
                 self.logger.error(f"Error in cleanup thread: {e}")
-                time.sleep(60)  # Wait a minute before retrying
+                if self._stop.wait(60):  # Wait a minute before retrying
+                    break
+
+    def shutdown(self, timeout: float = TRACKER_JOIN_SECONDS) -> bool:
+        """Stop the cleanup thread and wait for it within ``timeout`` seconds.
+
+        Returns True when the thread has ended (or never existed). The tracker
+        keeps working afterwards; only the background pruning of expired
+        entries stops. Safe to call more than once. Called on the cleanup
+        thread itself it only signals and returns False.
+        """
+        stop = getattr(self, "_stop", None)
+        thread = getattr(self, "cleanup_thread", None)
+        if stop is None or thread is None:
+            return True  # built without __init__: no thread was started
+        stop.set()
+        if thread is threading.current_thread():
+            self.logger.debug("shutdown called on the prompt cleanup thread")
+            return False
+        thread.join(timeout)
+        if thread.is_alive():
+            self.logger.warning(
+                f"prompt cleanup thread still running after {timeout} s"
+            )
+            return False
+        return True
 
     def get_active_prompts(self) -> Dict[str, Dict[str, Any]]:
         """

web/js/workspace/system-tagging.mjs

@@ -85,6 +85,9 @@ export const STATUS_COPY = Object.freeze({
 export const SETTING_PROBLEMS = Object.freeze({
     autotag_general_threshold: 'General-tag confidence must be between 0 and 1.',
     autotag_character_threshold: 'Character-tag confidence must be between 0 and 1.',
+    // #369: every page shows the thresholds in steps of 0.05, so the server saves only values on that grid.
+    autotag_general_threshold_step: 'General-tag confidence must be a multiple of 0.05.',
+    autotag_character_threshold_step: 'Character-tag confidence must be a multiple of 0.05.',
     autotag_suggestion_limit: 'Suggestions per image must be a whole number from 1 to 100.',
     // The server's two template sentences (`validate_template`), blank first.
     autotag_template: 'LLM tag template must be non-empty text.',
@@ -171,10 +174,15 @@ export function draftValue(key, text) {
 }
 
 const unit = value => typeof value === 'number' && Number.isFinite(value) && value >= 0 && value <= 1;
+// A multiple of 0.05 within the server's tolerance (`validate_saved_threshold`): (0.1 + 0.2) * 20 is 6.000000000000001.
+const onStep = value => Math.abs(value * 20 - Math.round(value * 20)) <= 1e-12;
+// A threshold on the grid, handed on exactly as the server stores it (anything else is left for draftProblem to refuse).
+const snapped = value => (unit(value) && onStep(value) ? Math.round(value * 20) / 20 : value);
 
 /**
  * The first setting the chosen model would use that the section cannot send, as `{key, message}`, else null: WD14
- * reads both thresholds, the language models the template; Generate also reads the suggestion count (`limit`).
+ * reads both thresholds (0 to 1 in steps of 0.05, #369), the language models the template; Generate also reads the
+ * suggestion count (`limit`).
  */
 export function draftProblem(values, { limit = true } = {}) {
     const model = values.autotag_model_type;
@@ -189,6 +197,7 @@ export function draftProblem(values, { limit = true } = {}) {
         }
         const valid = key === 'autotag_suggestion_limit' ? Number.isInteger(value) && value >= 1 && value <= 100 : unit(value);
         if (!valid) return { key, message: SETTING_PROBLEMS[key] };
+        if (key !== 'autotag_suggestion_limit' && !onStep(value)) return { key, message: SETTING_PROBLEMS[`${key}_step`] };
     }
     return null;
 }
@@ -197,7 +206,8 @@ export function draftProblem(values, { limit = true } = {}) {
 export function taggingPreset(values) {
     const model = values.autotag_model_type;
     const options = CLASSIFIERS.has(model)
-        ? { general_threshold: values.autotag_general_threshold, character_threshold: values.autotag_character_threshold }
+        ? { general_threshold: snapped(values.autotag_general_threshold),
+            character_threshold: snapped(values.autotag_character_threshold) }
         : { instructions: values.autotag_template };
     return { model, options: { ...options, suggestion_limit: values.autotag_suggestion_limit } };
 }
@@ -206,7 +216,8 @@ export function taggingPreset(values) {
 export function batchBody(values) {
     const model = values.autotag_model_type;
     const options = CLASSIFIERS.has(model)
-        ? { general_threshold: values.autotag_general_threshold, character_threshold: values.autotag_character_threshold }
+        ? { general_threshold: snapped(values.autotag_general_threshold),
+            character_threshold: snapped(values.autotag_character_threshold) }
         : { instructions: values.autotag_template };
     return { model_type: model, existing: values.autotag_existing, options };
 }