fix(restore): keep the cancel and the adoption when a stopped restore's publish ends (#360)

A publish that fails after the cancel now ends in CancelledError (after
its prune) instead of a 500, and the restored library is adopted to its
end even when a second cancel arrives. The blocking calls go through
_run_in_executor again, as the executor contract test expects.

Exit-window refusals: the comments state what was measured (a pool is
refused during threading._shutdown; Thread.start only once the
interpreter is finalizing), and the tests assert the one DEBUG line.

py/api/__init__.py

@@ -1107,196 +1107,3 @@ class PromptManagerAPI(
                         result["parameters"][field] = metadata[field]
 
         return result
-
-    def _extract_readable_prompt(self, parsed_data):
-        """Extract human-readable prompt text from ComfyUI/A1111 data."""
-
-        def safe_to_string(value):
-            if isinstance(value, str):
-                return value
-            elif isinstance(value, list):
-                return " ".join(str(item) for item in value if item)
-            elif value is not None:
-                return str(value)
-            return None
-
-        # First check if we already extracted a positive prompt (A1111 format)
-        if parsed_data.get("positive_prompt"):
-            return parsed_data["positive_prompt"]
-
-        # Check if prompt is already a string
-        if isinstance(parsed_data.get("prompt"), str):
-            return parsed_data["prompt"]
-
-        # Check if prompt is a simple value that can be converted
-        if parsed_data.get("prompt") and not isinstance(
-            parsed_data.get("prompt"), dict
-        ):
-            return safe_to_string(parsed_data["prompt"])
-
-        prompt_data = parsed_data.get("prompt")
-        if isinstance(prompt_data, dict):
-            # Use the enhanced logic from parse-metadata.py
-            positive_prompt = self._extract_positive_prompt_from_comfyui_data(
-                prompt_data
-            )
-            if positive_prompt:
-                return positive_prompt
-
-        # Check workflow data if available
-        workflow_data = parsed_data.get("workflow")
-        if isinstance(workflow_data, dict):
-            positive_prompt = self._extract_positive_prompt_from_comfyui_data(
-                workflow_data
-            )
-            if positive_prompt:
-                return positive_prompt
-
-        # Check parameters for positive prompt
-        if parsed_data.get("parameters", {}).get("positive"):
-            return safe_to_string(parsed_data["parameters"]["positive"])
-
-        return None
-
-    def _get_node_inputs(self, node):
-        """Safely get inputs from a node, handling both dict and list formats.
-
-        Old format: inputs is a dict with direct key-value pairs
-            inputs = {"text": "my prompt", "seed": 123}
-
-        New format: inputs is a list of connection objects
-            inputs = [
-                {"name": "text", "type": "STRING", "link": null,
-                 "widget": {"name": "text"}},
-                {"name": "clip", "type": "CLIP", "link": 11}
-            ]
-        """
-        if not isinstance(node, dict):
-            return {}
-
-        inputs = node.get("inputs", {})
-
-        # If inputs is already a dict, return it
-        if isinstance(inputs, dict):
-            return inputs
-
-        # If inputs is a list, convert to dict format
-        if isinstance(inputs, list):
-            inputs_dict = {}
-            for input_item in inputs:
-                if isinstance(input_item, dict) and "name" in input_item:
-                    name = input_item["name"]
-                    inputs_dict[name] = input_item
-            return inputs_dict
-
-        return {}
-
-    def _find_text_in_node(self, node):
-        """Try to find text content in a node using various strategies.
-
-        Handles both old and new workflow formats.
-        """
-        if not isinstance(node, dict):
-            return None
-
-        # Strategy 1: Check normalized inputs for 'text' field
-        inputs = self._get_node_inputs(node)
-        if "text" in inputs and isinstance(inputs["text"], str):
-            return inputs["text"]
-
-        # Strategy 2: For text encoder nodes, check widgets_values
-        class_type = node.get("class_type", node.get("type", ""))
-        text_encoder_types = [
-            "CLIPTextEncode",
-            "CLIPTextEncodeSDXL",
-            "CLIPTextEncodeSDXLRefiner",
-            "CLIPTextEncodeFlux",
-            "PromptManager",
-            "PromptManagerText",
-            "BNK_CLIPTextEncoder",
-            "Text Encoder",
-            "CLIP Text Encode",
-        ]
-
-        if any(
-            encoder_type.lower() in class_type.lower()
-            for encoder_type in text_encoder_types
-        ):
-            widgets_values = node.get("widgets_values", [])
-            if widgets_values and len(widgets_values) > 0:
-                if isinstance(widgets_values[0], str) and widgets_values[0].strip():
-                    return widgets_values[0]
-
-        return None
-
-    def _extract_positive_prompt_from_comfyui_data(self, data):
-        """Extract positive prompt from ComfyUI data.
-
-        Handles both old and new formats.
-        """
-        if not isinstance(data, dict):
-            return None
-
-        # Build nodes dictionary
-        nodes_by_id = {}
-        if "nodes" in data:
-            # Handle nodes array format
-            for node in data["nodes"]:
-                if isinstance(node, dict):
-                    nid = node.get("id")
-                    if nid is not None:
-                        nodes_by_id[nid] = node
-        else:
-            # Handle flat dictionary format (node_id -> node_data)
-            for nid_str, node in data.items():
-                try:
-                    nid = int(nid_str)
-                except (ValueError, TypeError):
-                    nid = nid_str
-                if isinstance(node, dict):
-                    if "id" in node:
-                        nid = node["id"]
-                    nodes_by_id[nid] = node
-
-        if not nodes_by_id:
-            return None
-
-        # First, try to find positive/negative connection pattern
-        pos_id = None
-        for node in nodes_by_id.values():
-            if isinstance(node, dict):
-                inputs = self._get_node_inputs(node)
-                if "positive" in inputs and "negative" in inputs:
-                    try:
-                        pos_input = inputs["positive"]
-                        if isinstance(pos_input, list) and len(pos_input) > 0:
-                            pos_id = int(pos_input[0])
-                            break
-                    except (ValueError, TypeError, IndexError):
-                        continue
-
-        # Get text from the positive node
-        if pos_id is not None and pos_id in nodes_by_id:
-            text_val = self._find_text_in_node(nodes_by_id[pos_id])
-            if text_val:
-                return text_val
-
-        # Fallback: find any text encoder node with text content
-        text_nodes = []
-        for node in nodes_by_id.values():
-            if isinstance(node, dict):
-                text_val = self._find_text_in_node(node)
-                if text_val:
-                    # Try to determine if this is positive or negative
-                    node_title = node.get("title", "").lower()
-                    if "neg" not in node_title and "negative" not in node_title:
-                        # Prioritize non-negative prompts
-                        text_nodes.insert(0, text_val)
-                    else:
-                        text_nodes.append(text_val)
-
-        # Return the first positive-looking prompt
-        if text_nodes:
-            return text_nodes[0]
-
-        return None

py/api/admin.py

@@ -47,7 +47,7 @@ try:
         extract_generation_params,
         parse_page_json,
     )
-    from ...utils.prompt_pair_resolver import resolve_prompt_pair
+    from ...utils.prompt_pair_resolver import resolve_prompt_pair, sampler_less_prompt
     from ...utils.review_capture import reviewable
 except ImportError:
     from database.pair_ingestion import blank_positive, ingest_resolved_pair
@@ -62,7 +62,7 @@ except ImportError:
         extract_generation_params,
         parse_page_json,
     )
-    from utils.prompt_pair_resolver import resolve_prompt_pair
+    from utils.prompt_pair_resolver import resolve_prompt_pair, sampler_less_prompt
     from utils.review_capture import reviewable
 
 
@@ -1754,13 +1754,24 @@ class AdminRoutesMixin:
             canonical.get("parameters"), str
         ):
             return resolution["state"]
+        elif canonical.get("prompt"):
+            # No node takes a positive or negative input (a guider or wrapper
+            # graph): the one wired text, never a guess (#252).
+            found = sampler_less_prompt(canonical["prompt"])
+            prompt_text = found["text"]
+            if prompt_text is None and isinstance(canonical.get("parameters"), str):
+                # A metadata saver also wrote A1111 text: its positive prompt.
+                parsed = self._parse_comfyui_prompt(metadata)["positive_prompt"]
+                prompt_text = parsed.strip() if parsed and parsed.strip() else None
+            if prompt_text is None:
+                linked = self._link_recorded_run(
+                    media_file, canonical, detailed, history
+                )
+                if linked is not None:
+                    return linked
+                return "unresolved" if found["state"] == "ambiguous" else "found"
         else:
-            parsed = self._parse_comfyui_prompt(metadata)
-            if not (parsed.get("prompt") or parsed.get("parameters")):
-                return None
-            prompt_text = self._extract_readable_prompt(parsed)
-            if prompt_text is not None and not isinstance(prompt_text, str):
-                prompt_text = str(prompt_text)
+            prompt_text = self._parse_comfyui_prompt(metadata)["positive_prompt"]
             if not (prompt_text and prompt_text.strip()):
                 return "found"
             prompt_text = prompt_text.strip()

py/api/backup_restore.py

@@ -483,13 +483,59 @@ class BackupRestoreRoutesMixin:
             self.logger.info(f"Removed {len(removed)} old safety backups")
 
     async def _publish(self, model, staged):
-        """Publish ``staged``; on a page-copy failure prune and re-raise."""
+        """Publish ``staged``; on a page-copy failure prune and re-raise.
+
+        A cancel (server stop) waits for the publishing thread (#360): the
+        caller's cleanup and its registry admission must outlive it. When the
+        copy then succeeded the library was replaced, so it is adopted
+        (``_after_publish``, itself run to its end) before the cancel goes
+        on; a copy that failed is handled as usual, then the cancel goes on.
+        """
+        future = self._in_executor_task(model.publish_staged, staged)
+        cancelled = await _outlive_cancel(future)
         try:
-            return await self._run_in_executor(model.publish_staged, staged)
+            backup_path = future.result()
         except sqlite3.OperationalError as exc:
             if getattr(exc, "_prompt_manager_restore_backup_created", False) is True:
                 await self._prune_after(model.db_path, failed=True)
+            if cancelled:
+                raise asyncio.CancelledError from exc
             raise
+        except Exception as exc:
+            if cancelled:
+                raise asyncio.CancelledError from exc
+            raise
+        if cancelled:
+            self.logger.info("The restore finished while the server was stopping")
+            # A second cancel (the loop's own shutdown) must not cut it short.
+            adopted = asyncio.ensure_future(self._after_publish(model.db_path))
+            await _outlive_cancel(adopted)
+            if not adopted.cancelled() and adopted.exception() is not None:
+                self.logger.error(
+                    "Could not adopt the restored library",
+                    exc_info=adopted.exception(),
+                )
+            raise asyncio.CancelledError
+        return backup_path
+
+    async def _stage_to_end(self, model, temp_path):
+        """Stage ``temp_path``; a cancel waits for the worker (#360).
+
+        The upload stays until the staging thread is done with it, and a
+        stage it finished after the cancel is removed before the cancel goes
+        on.
+        """
+        future = self._in_executor_task(model.stage_restore, temp_path)
+        if await _outlive_cancel(future):
+            if not future.cancelled() and future.exception() is None:
+                await self._discard_held([_Orphan(future.result())])
+            raise asyncio.CancelledError
+        return future.result()
+
+    def _in_executor_task(self, func, *args):
+        """``_run_in_executor(func, *args)`` as a task ``_outlive_cancel`` can
+        wait for (shielded, the executor call is never cancelled)."""
+        return asyncio.ensure_future(self._run_in_executor(func, *args))
 
     async def _after_publish(self, db_path):
         """Everything after the copy replaced the library: best effort, logged.
@@ -609,7 +655,7 @@ class BackupRestoreRoutesMixin:
                     {"success": False, "error": REQUESTS_BUSY}, status=409
                 )
             model = self.db.model
-            staged = await self._run_in_executor(model.stage_restore, temp_path)
+            staged = await self._stage_to_end(model, temp_path)
             try:
                 backup_path = await self._publish(model, staged)
             finally:
@@ -927,6 +973,30 @@ class BackupRestoreRoutesMixin:
         job.update(progress=100, status_line=line)
 
 
+async def _outlive_cancel(future):
+    """Await ``future`` (an executor call's task) to its end, even through
+    cancels (#360).
+
+    A cancelled task cannot stop the worker thread, so the files it uses and
+    the registry admission it runs under must outlive it: the cancel is held
+    until the worker returned. Returns whether the task was cancelled
+    meanwhile; the caller reads the outcome from ``future`` and re-raises
+    ``CancelledError``. A future cancelled itself (an executor shut down) is
+    not a cancel of the task: ``future.result()`` raises it. Never catches
+    ``GeneratorExit``: a closed loop still closes the coroutine.
+    """
+    cancelled = False
+    while not future.done():
+        try:
+            await asyncio.shield(future)
+        except asyncio.CancelledError:
+            if not future.cancelled():
+                cancelled = True
+        except Exception:
+            pass  # the caller reads it from the future
+    return cancelled
+
+
 class _Orphan:
     """A stage with no store entry, in the shape ``_discard_held`` takes."""
 

py/api/thumbnail_auto.py

@@ -88,9 +88,9 @@ from .thumbnail_encode import shutdown_encoders, writer_settings
 from .thumbnail_jobs import ThumbnailOptions
 
 try:
-    from ...utils.parallel import map_parallel
+    from ...utils.parallel import map_parallel, note_refusal, refused_at_shutdown
 except ImportError:  # tests import the package top-level
-    from utils.parallel import map_parallel
+    from utils.parallel import map_parallel, note_refusal, refused_at_shutdown
 
 # The placeholder frame a video without a preview is answered with (JPEG
 # bytes, never written to disk); written previews follow the settings.
@@ -410,7 +410,8 @@ class AutoThumbnailer:
 
     def enqueue(self, src, dst, is_video):
         """Queue *dst*'s preview for the worker; False when already queued,
-        full or closed. Never blocks on a write."""
+        full, closed or its worker is refused at interpreter exit. Never
+        blocks on a write."""
         key = _dst_key(dst)
         with self._changed:
             if self._closed or key in self._queued:
@@ -420,30 +421,41 @@ class AutoThumbnailer:
             self._queued.add(key)
             self._queue.append((key, Path(src), Path(dst), bool(is_video)))
             self._busy = True
+            started = True
             if self._start_worker and not self._worker_running:
-                self._start_worker_locked()
+                started = self._start_worker_locked()
+            if not started:
+                # Refused at interpreter exit: no worker will take it (#375).
+                self._queue.pop()
+                self._queued.discard(key)
             self._changed.notify_all()
-        return True
+        return started
 
     def _start_worker_locked(self):
-        """Start the worker (lock held). A start that fails leaves the file
-        queued and the next enqueue tries again."""
+        """Start the worker (lock held); False only when the interpreter is
+        exiting (#375: logged once at DEBUG, the caller drops the file).
+        Another start failure leaves the file queued and the next enqueue
+        tries again."""
         worker = threading.Thread(target=self._run, name=WORKER_NAME, daemon=True)
         worker.pm_preview_token = _MODULE_TOKEN
         worker.pm_preview_close = self.close
         self._worker_running = True
         try:
             worker.start()
-        except RuntimeError:
+        except RuntimeError as error:
             self._worker_running = False
             self._busy = False
+            if refused_at_shutdown(error):
+                note_refusal(self._logger, "The automatic preview worker")
+                return False
             self._logger.warning(
                 "The automatic preview worker could not start; the next output "
                 "tries again",
                 exc_info=True,
             )
-            return
+            return True
         self._worker = worker
+        return True
 
     def _run(self):
         try:
@@ -790,8 +802,8 @@ class AutoThumbnailMixin:
 
         Returns ``queued``, ``fresh`` (its preview is current) or ``skipped``
         (not a media file under a current root, a link, a preview itself,
-        already queued, or after the app's cleanup). Never writes on the
-        caller's thread.
+        already queued, after the app's cleanup, or when its worker is refused
+        at interpreter exit). Never writes on the caller's thread.
         """
         if self._auto_thumbnails_shut:
             return "skipped"

tests/test_api_admin.py

@@ -846,6 +846,14 @@ class TestSettingsReadMonitorThroughModule(AdminAPITestCase):
         self.assertEqual(data["settings"]["gallery_root_paths"], ["output"])
 
 
+def _wired_prompt(text):
+    """A sampler-less graph whose one encoder feeds a guider (#252)."""
+    return {
+        "1": {"class_type": "CLIPTextEncode", "inputs": {"text": text}},
+        "2": {"class_type": "BasicGuider", "inputs": {"conditioning": ["1", 0]}},
+    }
+
+
 def _raise(*args, **kwargs):
     raise RuntimeError("boom")
 
@@ -2186,11 +2194,7 @@ class TestOutputScanInternals(AdminAPITestCase):
 
     def test_ingest_outcomes(self):
         f = Path("gen.png")
-        meta = {
-            "prompt": json.dumps(
-                {"1": {"class_type": "CLIPTextEncode", "inputs": {"text": "a cat"}}}
-            )
-        }
+        meta = {"prompt": json.dumps(_wired_prompt("a cat"))}
         self.assertIsNone(self.api._ingest_scanned_file(f, {}))
         self.assertIsNone(self.api._ingest_scanned_file(f, {"prompt": "{}"}))
         self.assertEqual(
@@ -2202,31 +2206,19 @@ class TestOutputScanInternals(AdminAPITestCase):
 
     def test_ingest_link_failures_are_tolerated(self):
         f = Path("gen.png")
-        meta = {
-            "prompt": json.dumps(
-                {"1": {"class_type": "CLIPTextEncode", "inputs": {"text": "a dog"}}}
-            )
-        }
+        meta = {"prompt": json.dumps(_wired_prompt("a dog"))}
         self.api.db.link_image_to_prompt = _raise
         self.assertEqual(self.api._ingest_scanned_file(f, meta), "added")
         self.assertEqual(self.api._ingest_scanned_file(f, meta), "found")
 
     def test_ingest_save_failure_counts_as_found(self):
         f = Path("gen.png")
-        meta = {
-            "prompt": json.dumps(
-                {"1": {"class_type": "CLIPTextEncode", "inputs": {"text": "a bird"}}}
-            )
-        }
+        meta = {"prompt": json.dumps(_wired_prompt("a bird"))}
         self.api.db.save_prompt = lambda *a, **k: None
         self.assertEqual(self.api._ingest_scanned_file(f, meta), "found")
 
     def test_batch_counts_and_per_file_errors(self):
-        meta = {
-            "prompt": json.dumps(
-                {"1": {"class_type": "CLIPTextEncode", "inputs": {"text": "a fish"}}}
-            )
-        }
+        meta = {"prompt": json.dumps(_wired_prompt("a fish"))}
         self.api.db.get_prompt_by_hash = _raise
         counts = self.api._ingest_scan_batch_sync(
             [(Path("a.png"), meta), (Path("b.png"), {})]
@@ -2915,7 +2907,7 @@ class TestScanReadsVideos(AdminAPITestCase):
         shutil.which("ffmpeg") and shutil.which("ffprobe"), "ffmpeg not installed"
     )
     async def test_scan_adds_prompts_found_in_videos(self):
-        graph = {"1": {"class_type": "CLIPTextEncode", "inputs": {"text": "a cat"}}}
+        graph = _wired_prompt("a cat")
         _write_mp4(
             self.output_dir / "clip.mp4", json.dumps({"prompt": json.dumps(graph)})
         )

tests/test_api_core.py

@@ -835,7 +835,6 @@ class TestMetadataParsing(unittest.TestCase):
         self.assertEqual(parsed["positive_prompt"], "a cat")
         self.assertEqual(parsed["negative_prompt"], "dog")
         self.assertEqual(parsed["parameters"]["parameters"], meta["parameters"])
-        self.assertEqual(self.api._extract_readable_prompt(parsed), "a cat")
 
     def test_parse_comfyui_fields(self):
         meta = {
@@ -856,97 +855,6 @@ class TestMetadataParsing(unittest.TestCase):
         self.assertEqual(parsed["prompt"], "plain text")
         self.assertEqual(parsed["workflow"], {"nodes": []})
 
-    def test_extract_readable_prompt_variants(self):
-        extract = self.api._extract_readable_prompt
-        self.assertEqual(extract({"prompt": "plain"}), "plain")
-        self.assertEqual(extract({"prompt": ["a", "", "b"]}), "a b")
-        self.assertEqual(extract({"prompt": 42}), "42")
-        self.assertEqual(
-            extract({"prompt": None, "parameters": {"positive": ["x", "y"]}}), "x y"
-        )
-        self.assertIsNone(extract({"prompt": {}, "parameters": {}}))
-
-        graph = {
-            "1": {"class_type": "CLIPTextEncode", "inputs": {"text": "pos"}},
-            "2": {"class_type": "CLIPTextEncode", "inputs": {"text": "neg"}},
-            "3": {
-                "class_type": "KSampler",
-                "inputs": {"positive": ["1", 0], "negative": ["2", 0]},
-            },
-        }
-        self.assertEqual(extract({"prompt": graph}), "pos")
-        self.assertEqual(extract({"prompt": {"x": 1}, "workflow": graph}), "pos")
-
-    def test_node_inputs_and_text_lookup(self):
-        self.assertEqual(self.api._get_node_inputs("nope"), {})
-        self.assertEqual(self.api._get_node_inputs({"inputs": 5}), {})
-        self.assertEqual(self.api._get_node_inputs({"inputs": {"a": 1}}), {"a": 1})
-        listed = self.api._get_node_inputs(
-            {"inputs": [{"name": "text", "link": None}, {"nope": 1}, "junk"]}
-        )
-        self.assertEqual(list(listed), ["text"])
-
-        self.assertIsNone(self.api._find_text_in_node(None))
-        self.assertEqual(self.api._find_text_in_node({"inputs": {"text": "hi"}}), "hi")
-        self.assertEqual(
-            self.api._find_text_in_node(
-                {"type": "CLIPTextEncode", "widgets_values": ["w"]}
-            ),
-            "w",
-        )
-        self.assertIsNone(
-            self.api._find_text_in_node(
-                {"type": "CLIPTextEncode", "widgets_values": [" "]}
-            )
-        )
-        self.assertIsNone(self.api._find_text_in_node({"type": "KSampler"}))
-
-    def test_positive_prompt_from_nodes_array_and_fallbacks(self):
-        extract = self.api._extract_positive_prompt_from_comfyui_data
-        self.assertIsNone(extract("x"))
-        self.assertIsNone(extract({}))
-        self.assertIsNone(extract({"nodes": []}))
-
-        workflow = {
-            "nodes": [
-                {"id": 1, "type": "CLIPTextEncode", "widgets_values": ["pos"]},
-                {
-                    "id": 2,
-                    "type": "CLIPTextEncode",
-                    "widgets_values": ["neg"],
-                    "title": "Negative",
-                },
-                {"id": 3, "type": "KSampler", "inputs": [{"name": "seed", "link": 4}]},
-                "junk",
-            ]
-        }
-        self.assertEqual(extract(workflow), "pos")
-
-        only_negative = {
-            "nodes": [
-                {
-                    "id": 2,
-                    "type": "CLIPTextEncode",
-                    "widgets_values": ["neg"],
-                    "title": "neg",
-                },
-            ]
-        }
-        self.assertEqual(extract(only_negative), "neg")
-
-        bad_link = {
-            "1": {"class_type": "CLIPTextEncode", "inputs": {"text": "pos"}},
-            "3": {
-                "class_type": "KSampler",
-                "inputs": {"positive": "oops", "negative": []},
-            },
-            "4": {
-                "class_type": "KSampler",
-                "inputs": {"positive": ["zz", 0], "negative": []},
-            },
-        }
-        self.assertEqual(extract(bad_link), "pos")
-
 
 if __name__ == "__main__":
     unittest.main()

tests/test_exit_window_threads.py

@@ -0,0 +1,266 @@
+"""A thread or pool refused at interpreter shutdown is quiet (#375).
+
+When the interpreter exits, ``threading._shutdown`` first shuts every
+``ThreadPoolExecutor`` (``submit`` then raises "cannot schedule new futures
+after interpreter shutdown") and joins the non-daemon threads; only then do
+the ``atexit`` callbacks (the package teardown) run. Our daemon threads keep
+working meanwhile. ``Thread.start`` is refused ("can't create new thread at
+interpreter shutdown") only later, once the interpreter is finalizing; the
+unit tests cover that case, the child process the pool case. Such a refusal
+is logged once at DEBUG, with no traceback, and leaves nothing pending.
+
+``threading.Thread`` is replaced only in the module under test (a copy of
+the ``threading`` namespace), never process-wide. The child-process test runs
+the real exit: a daemon thread queues previews while the main thread returns.
+"""
+
+import json
+import logging
+import os
+import subprocess
+import sys
+import tempfile
+import textwrap
+import threading
+import types
+import unittest
+from concurrent.futures import ThreadPoolExecutor
+from pathlib import Path
+from unittest import mock
+
+from py.api import thumbnail_auto
+from utils import image_monitor, parallel
+
+ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+CHILD_CEILING_SECONDS = 60  # the hang watchdog allows 120 s per test
+REPORT_PREFIX = "PM_EXIT_REPORT "
+THREAD_REFUSAL = "can't create new thread at interpreter shutdown"
+POOL_REFUSAL = "cannot schedule new futures after interpreter shutdown"
+
+
+def _refusing_threading():
+    """A ``threading`` namespace whose ``Thread.start`` is refused."""
+
+    class RefusedThread(threading.Thread):
+        def start(self):
+            raise RuntimeError(THREAD_REFUSAL)
+
+    namespace = types.SimpleNamespace(**vars(threading))
+    namespace.Thread = RefusedThread
+    return namespace
+
+
+class _ShutPool(ThreadPoolExecutor):
+    def submit(self, *args, **kwargs):
+        raise RuntimeError(POOL_REFUSAL)
+
+
+class _Capture(logging.Handler):
+    def __init__(self):
+        super().__init__(logging.DEBUG)
+        self.records = []
+
+    def emit(self, record):
+        self.records.append(record)
+
+
+def _capture(testcase, name):
+    parallel._noted.clear()  # "once" is per process: start each test afresh
+    logger = logging.getLogger(name)
+    handler = _Capture()
+    previous = logger.level
+    logger.addHandler(handler)
+    logger.setLevel(logging.DEBUG)
+
+    def restore():
+        logger.removeHandler(handler)
+        logger.setLevel(previous)
+
+    testcase.addCleanup(restore)
+    return handler.records
+
+
+def _assert_one_debug_line(testcase, records):
+    """Exactly one refusal line, at DEBUG and without a traceback."""
+    refusals = [r for r in records if "interpreter is exiting" in r.getMessage()]
+    testcase.assertEqual(len(refusals), 1, [r.getMessage() for r in records])
+    testcase.assertEqual(refusals[0].levelno, logging.DEBUG)
+    testcase.assertEqual([r for r in records if r.levelno > logging.DEBUG], [])
+    testcase.assertTrue(all(r.exc_info is None for r in records))
+
+
+class TestScheduleRefusedAtShutdown(unittest.TestCase):
+    def setUp(self):
+        self.handler = image_monitor.ImageGenerationHandler(
+            mock.MagicMock(), mock.MagicMock()
+        )
+        self.addCleanup(self.handler.close)
+
+    def test_schedule_returns_false_and_leaves_nothing_pending(self):
+        records = _capture(self, "prompt_manager.image_monitor")
+        with mock.patch.object(image_monitor, "threading", _refusing_threading()):
+            self.assertFalse(self.handler.schedule("/out/a.png"))
+            self.assertFalse(self.handler.schedule("/out/b.png"))
+        self.assertEqual(self.handler.pending_count(), 0)
+        _assert_one_debug_line(self, records)
+
+    def test_another_start_failure_still_raises_and_leaves_nothing_pending(self):
+        class Failing(threading.Thread):
+            def start(self):
+                raise RuntimeError("can't start new thread")
+
+        namespace = types.SimpleNamespace(**vars(threading))
+        namespace.Thread = Failing
+        with mock.patch.object(image_monitor, "threading", namespace):
+            with self.assertRaises(RuntimeError):
+                self.handler.schedule("/out/a.png")
+        self.assertEqual(self.handler.pending_count(), 0)
+
+    def test_schedule_before_shutdown_is_unchanged(self):
+        self.handler.processing_delay = 3600
+        self.assertTrue(self.handler.schedule("/out/a.png"))
+        self.assertEqual(self.handler.pending_count(), 1)
+
+
+class TestPreviewQueueRefusedAtShutdown(unittest.TestCase):
+    def thumbnailer(self, logger):
+        made = thumbnail_auto.AutoThumbnailer(
+            lambda *a, **k: {"action": "generated"}, lambda: True, logger
+        )
+        self.addCleanup(made.close)
+        return made
+
+    def test_enqueue_is_refused_quietly_and_drops_the_file(self):
+        logger = logging.getLogger("prompt_manager.test_exit_window")
+        records = _capture(self, logger.name)
+        thumbnailer = self.thumbnailer(logger)
+        src = Path("/out/a.png")
+        with mock.patch.object(thumbnail_auto, "threading", _refusing_threading()):
+            self.assertFalse(thumbnailer.enqueue(src, Path("/out/t/a.webp"), False))
+            self.assertFalse(thumbnailer.enqueue(src, Path("/out/t/b.webp"), False))
+        self.assertFalse(thumbnailer._worker_running)
+        self.assertEqual(len(thumbnailer._queue), 0)
+        self.assertEqual(thumbnailer._queued, set())
+        self.assertTrue(thumbnailer.wait_idle(0))
+        _assert_one_debug_line(self, records)
+
+
+class TestMapParallelRefusedAtShutdown(unittest.TestCase):
+    def test_a_shut_pool_runs_the_items_on_the_calling_thread(self):
+        records = _capture(self, "prompt_manager.parallel")
+        with mock.patch.object(parallel, "ThreadPoolExecutor", _ShutPool):
+            self.assertEqual(
+                parallel.map_parallel(lambda n: n * n, [1, 2, 3], 4), [1, 4, 9]
+            )
+        _assert_one_debug_line(self, records)
+
+    def test_another_pool_error_still_raises(self):
+        class Broken(ThreadPoolExecutor):
+            def submit(self, *args, **kwargs):
+                raise RuntimeError("something else")
+
+        with mock.patch.object(parallel, "ThreadPoolExecutor", Broken):
+            with self.assertRaises(RuntimeError):
+                parallel.map_parallel(lambda n: n, [1, 2], 2)
+
+
+CHILD = textwrap.dedent("""
+    import threading
+
+    # Registered before concurrent.futures is imported, so it runs after the
+    # pool's own exit hook: from here on, a pool refuses new work.
+    exiting = threading.Event()
+    threading._register_atexit(exiting.set)
+
+    import json
+    import logging
+    import os
+    import sys
+    from pathlib import Path
+
+    root, tmp = sys.argv[1], sys.argv[2]
+    sys.path.insert(0, root)
+
+    from py.api.thumbnail_auto import AutoThumbnailer
+
+    logging.basicConfig(level=logging.WARNING, stream=sys.stderr)
+    logger = logging.getLogger("prompt_manager.exit_child")
+
+    def generate(src, dst, is_video, *, stop, guard, track):
+        Path(dst).write_bytes(b"preview")
+        return {"action": "generated"}
+
+    thumbnailer = AutoThumbnailer(
+        generate, lambda: True, logger, workers=lambda: 4
+    )
+    queued = threading.Event()
+
+    def late_monitor():
+        exiting.wait()
+        # One batch of two: the worker maps it on a pool.
+        with thumbnailer._changed:
+            for name in ("a", "b"):
+                src = Path(tmp, name + ".png")
+                src.write_bytes(b"x")
+                thumbnailer.enqueue(src, Path(tmp, name + "_thumb.webp"), False)
+        queued.set()
+
+    def holder():
+        # Non-daemon: the interpreter waits for it, inside the exit window.
+        exiting.wait()
+        queued.wait(30)
+        idle = thumbnailer.wait_idle(30)
+        written = sorted(
+            n for n in os.listdir(tmp) if n.endswith("_thumb.webp")
+        )
+        print("PM_EXIT_REPORT " + json.dumps({"idle": idle, "written": written}))
+        sys.stdout.flush()
+
+    threading.Thread(target=late_monitor, daemon=True).start()
+    threading.Thread(target=holder).start()
+    # Return normally, as ComfyUI does after Ctrl+C.
+    """)
+
+
+class TestExitWindowChild(unittest.TestCase):
+    def test_previews_queued_in_the_exit_window_log_no_traceback(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            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)
+            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}"
+                )
+        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])
+        self.assertTrue(report["idle"], detail)
+        # The batch still ran: on the worker's own thread.
+        self.assertEqual(report["written"], ["a_thumb.webp", "b_thumb.webp"], detail)
+        self.assertNotIn("Traceback", completed.stderr, detail)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_pair_scan.py

@@ -115,7 +115,12 @@ class TestPairScan(AdminAPITestCase):
         self.assertEqual(second["pairs_added"], 0)
 
     def test_legacy_single_prompt_second_pass_not_reported_as_new_link(self):
-        meta = {"prompt": json.dumps({"1": graph()["1"]})}
+        # A sampler-less graph: one encoder wired into a guider (#252).
+        single = {
+            "1": graph()["1"],
+            "2": {"class_type": "BasicGuider", "inputs": {"conditioning": ["1", 0]}},
+        }
+        meta = {"prompt": json.dumps(single)}
         path = self.output_dir / "old.png"
         self.assertEqual(self.api._ingest_scanned_file(path, meta), "added")
         self.assertEqual(self.api._ingest_scanned_file(path, meta), "unchanged")
@@ -292,3 +297,155 @@ class TestPairScan(AdminAPITestCase):
             self.api.db.get_image_prompt_info(str(path))["workflow_data"], workflow
         )
         self.assertEqual(self.api._ingest_scanned_file(path, metadata), "unchanged")
+
+
+def flux_graph(text="a red fox", encoder="CLIPTextEncode"):
+    """A sampler-less Flux graph: the prompt reaches a guider's conditioning (#252)."""
+    return {
+        "1": {"class_type": "UNETLoader", "inputs": {"unet_name": "flux.safetensors"}},
+        "2": {"class_type": "DualCLIPLoader", "inputs": {"clip_name1": "t5"}},
+        "6": {"class_type": encoder, "inputs": {"text": text, "clip": ["2", 0]}},
+        "22": {
+            "class_type": "BasicGuider",
+            "inputs": {"model": ["1", 0], "conditioning": ["6", 0]},
+        },
+        "13": {
+            "class_type": "SamplerCustomAdvanced",
+            "inputs": {"guider": ["22", 0], "sigmas": ["17", 0]},
+        },
+        "17": {"class_type": "BasicScheduler", "inputs": {"steps": 20}},
+        "9": {"class_type": "SaveImage", "inputs": {"images": ["13", 0]}},
+    }
+
+
+class TestSamplerLessScan(AdminAPITestCase):
+    """Graphs where no node takes a positive or negative input (#252)."""
+
+    def _scanned(self, name, prompt_graph, workflow=None):
+        meta = {"prompt": json.dumps(prompt_graph)}
+        if workflow is not None:
+            meta["workflow"] = json.dumps(workflow)
+        return self.api._ingest_scanned_file(self.output_dir / name, meta)
+
+    def _texts(self):
+        return [p["text"] for p in self.api.db.get_recent_prompts()["prompts"]]
+
+    def test_the_one_wired_prompt_is_linked(self):
+        self.assertEqual(self._scanned("flux.png", flux_graph()), "added")
+        self.assertEqual(self._texts(), ["a red fox"])
+
+    def test_a_text_node_feeding_the_encoder_is_one_prompt(self):
+        g = flux_graph(text=["30", 0])
+        g["30"] = {"class_type": "TextInput", "inputs": {"text": "a red fox"}}
+        self.assertEqual(self._scanned("textinput.png", g), "added")
+        self.assertEqual(self._texts(), ["a red fox"])
+
+    def test_a_display_node_that_feeds_nothing_is_never_the_prompt(self):
+        # A real scan stored a "=== GIT CLONE REPORT ===" display text this way.
+        g = flux_graph()
+        g["40"] = {
+            "class_type": "easy showAnything",
+            "inputs": {"text": "=== GIT CLONE REPORT ===\nTotal repositories: 1"},
+        }
+        self.assertEqual(self._scanned("report.png", g), "added")
+        self.assertEqual(self._texts(), ["a red fox"])
+
+    def test_only_unconsumed_texts_store_nothing(self):
+        g = {
+            "40": {"class_type": "easy showAnything", "inputs": {"text": "report"}},
+            "41": {"class_type": "Textbox", "inputs": {"text": "install note"}},
+        }
+        self.assertEqual(self._scanned("notes.png", g), "found")
+        self.assertEqual(self._texts(), [])
+
+    def test_two_wired_texts_are_ambiguous_and_stay_unlinked(self):
+        # A LoRA loader's text input is wired too: the scan stored "<lora:...>".
+        g = flux_graph()
+        g["50"] = {
+            "class_type": "Lora Loader (LoraManager)",
+            "inputs": {
+                "model": ["1", 0],
+                "clip": ["2", 0],
+                "text": "<lora:style:1.00>",
+            },
+        }
+        g["22"]["inputs"]["model"] = ["50", 0]
+        g["6"]["inputs"]["clip"] = ["50", 1]
+        self.assertEqual(self._scanned("lora.png", g), "unresolved")
+        self.assertEqual(self._texts(), [])
+
+    def test_an_untraceable_wired_text_makes_the_graph_ambiguous(self):
+        # The encoder's text comes from a node no rule reads: never guess the other.
+        g = flux_graph(text=["60", 0])
+        g["60"] = {"class_type": "SomeLLMNode", "inputs": {"seed": 1}}
+        g["50"] = {
+            "class_type": "Lora Loader (LoraManager)",
+            "inputs": {"model": ["1", 0], "text": "<lora:style:1.00>"},
+        }
+        g["22"]["inputs"]["model"] = ["50", 0]
+        self.assertEqual(self._scanned("llm.png", g), "unresolved")
+        self.assertEqual(self._texts(), [])
+
+    def test_this_repositorys_prompt_node_is_authoritative(self):
+        g = flux_graph(text="a red fox", encoder="PromptManager")
+        g["50"] = {
+            "class_type": "Lora Loader (LoraManager)",
+            "inputs": {"model": ["1", 0], "text": "<lora:style:1.00>"},
+        }
+        g["22"]["inputs"]["model"] = ["50", 0]
+        self.assertEqual(self._scanned("own.png", g), "added")
+        self.assertEqual(self._texts(), ["a red fox"])
+
+    def test_a_numeric_text_is_never_stored_as_a_prompt(self):
+        self.assertEqual(self._scanned("number.png", flux_graph(text="2048")), "found")
+        self.assertEqual(self._texts(), [])
+
+    def test_workflow_widget_values_are_never_read(self):
+        # The API text is linked from a node no rule reads; the UI workflow
+        # still holds a stale widget value, which the old scan stored.
+        g = flux_graph(text=["70", 0])
+        g["70"] = {"class_type": "PromptBuilder", "inputs": {"prompt": "x", "n": 2}}
+        workflow = {
+            "nodes": [
+                {"id": 6, "type": "CLIPTextEncode", "widgets_values": ["stale text"]}
+            ]
+        }
+        self.assertEqual(self._scanned("stale.png", g, workflow), "unresolved")
+        self.assertEqual(self._texts(), [])
+
+    def test_a_lora_loaders_trigger_word_output_is_not_its_text(self):
+        # The encoder reads the loader's trigger-word output (slot 2), not its
+        # text input: the prompt is unknown, never the "<lora:...>" syntax.
+        g = flux_graph(text=["50", 2])
+        g["50"] = {
+            "class_type": "Lora Loader (LoraManager)",
+            "inputs": {"model": ["1", 0], "text": "<lora:style:1.00>"},
+        }
+        g["22"]["inputs"]["model"] = ["50", 0]
+        self.assertEqual(self._scanned("trigger.png", g), "unresolved")
+        self.assertEqual(self._texts(), [])
+
+    def test_a_seeded_text_node_holds_a_template_not_the_prompt(self):
+        g = flux_graph(text=["80", 0])
+        g["80"] = {
+            "class_type": "RandomPrompt",
+            "inputs": {"text": "a {red|blue} fox", "seed": 5},
+        }
+        self.assertEqual(self._scanned("random.png", g), "unresolved")
+        self.assertEqual(self._texts(), [])
+
+    def test_a1111_parameters_beside_an_undecided_graph_are_read(self):
+        # Metadata savers write both; the old scan read the A1111 positive.
+        g = flux_graph()
+        g["50"] = {
+            "class_type": "Lora Loader (LoraManager)",
+            "inputs": {"model": ["1", 0], "text": "<lora:style:1.00>"},
+        }
+        g["22"]["inputs"]["model"] = ["50", 0]
+        meta = {
+            "prompt": json.dumps(g),
+            "parameters": "a red fox\nNegative prompt: blur\nSteps: 20",
+        }
+        path = self.output_dir / "saver.png"
+        self.assertEqual(self.api._ingest_scanned_file(path, meta), "added")
+        self.assertEqual(self._texts(), ["a red fox"])

tests/test_restore_shutdown.py

@@ -0,0 +1,233 @@
+"""A restore cut off by a server stop waits for its worker thread (#360).
+
+A cancelled handler (or job) task cannot stop the executor thread that
+stages or publishes. Its cleanup (the stage, the upload) and the registry's
+``restore`` admission must outlive that thread, and the live library ends as
+the old or the restored one, never partial, with the safety backup kept.
+
+Every positive wait is on an event. The one bounded wait (``SETTLE``) is a
+negative check: the cancelled task must still be pending while its worker is
+held.
+"""
+
+import asyncio
+import os
+import shutil
+import sqlite3
+import threading
+from unittest.mock import patch
+
+from database import restore_stage
+from database.models import PromptModel
+from py.api import admin as admin_mod
+from py.api.jobs import RESTORE_JOB_KIND
+
+try:
+    from tests.test_restore_plans import RestoreTestCase, _texts
+except ImportError:  # discovered with tests/ as the top-level directory
+    from test_restore_plans import RestoreTestCase, _texts
+
+WAIT = 60  # a ceiling only a hung test reaches
+SETTLE = 0.3  # the negative check's bound: the old code's cleanup takes ms
+
+
+class _Gate:
+    """Hold one blocking call on its worker thread until released."""
+
+    def __init__(self, real):
+        self.real = real
+        self.entered = threading.Event()
+        self.release = threading.Event()
+        self.done = threading.Event()
+
+    def method(self):
+        """A function to patch onto the class (bound like the real one)."""
+
+        def held(*args):
+            return self(*args)
+
+        return held
+
+    def __call__(self, *args):
+        self.entered.set()
+        self.release.wait(WAIT)
+        try:
+            return self.real(*args)
+        finally:
+            self.done.set()
+
+
+class ShutdownRestoreCase(RestoreTestCase):
+    def gate(self, real):
+        gate = _Gate(real)
+        # A failed check never leaves the worker held until WAIT.
+        self._gates = getattr(self, "_gates", []) + [gate]
+        self.addCleanup(gate.release.set)
+        return gate
+
+    def stages(self):
+        """The stage folders (the upload's prefix starts the same way)."""
+        return [
+            name
+            for name in super().stages()
+            if os.path.isdir(os.path.join(self.tmpdir, name))
+        ]
+
+    def uploads(self):
+        """The upload files (not the -wal/-shm its readers may add)."""
+        return sorted(
+            name
+            for name in os.listdir(self.tmpdir)
+            if name.startswith(restore_stage.UPLOAD_PREFIX) and name.endswith(".tmp")
+        )
+
+    def fake_upload(self, texts):
+        """Make ``_receive_restore_upload`` hand over a prepared upload."""
+        self.upload(texts, "source.db")
+        source = os.path.join(self.tmpdir, "uploads", "source.db")
+        target = os.path.join(self.tmpdir, restore_stage.UPLOAD_PREFIX + "x.tmp")
+        shutil.copyfile(source, target)
+
+        async def receive(request, uploads):
+            uploads.append(target)
+            return target, None
+
+        return patch.object(self.api, "_receive_restore_upload", receive)
+
+    async def wait_thread(self, event):
+        loop = asyncio.get_running_loop()
+        self.assertTrue(await loop.run_in_executor(None, event.wait, WAIT))
+
+    async def assert_still_pending(self, task):
+        done, _ = await asyncio.wait({task}, timeout=SETTLE)
+        if done:
+            for gate in getattr(self, "_gates", ()):
+                gate.release.set()  # let the worker end before teardown
+            self.fail("the cancelled task ended while its worker still ran")
+
+
+class TestLegacyRestoreAtShutdown(ShutdownRestoreCase):
+    async def test_cancel_during_publish_waits_for_the_worker(self):
+        self.save("old")
+        old_db = self.api.db
+        gate = self.gate(PromptModel.publish_staged)
+        with (
+            self.fake_upload(["restored"]),
+            patch.object(PromptModel, "publish_staged", gate.method()),
+        ):
+            task = asyncio.ensure_future(self.api.restore_database(None))
+            await self.wait_thread(gate.entered)
+            self.assertEqual(len(self.stages()), 1)
+            task.cancel()
+            await self.assert_still_pending(task)
+            # The worker still has its stage and upload, and runs alone.
+            self.assertEqual(len(self.stages()), 1)
+            self.assertEqual(len(self.uploads()), 1)
+            self.assertIsNotNone(self.api._jobs.active(RESTORE_JOB_KIND))
+            gate.release.set()
+            with self.assertRaises(asyncio.CancelledError):
+                await asyncio.wait_for(task, WAIT)
+        self.assertTrue(gate.done.is_set())
+        self.assertEqual(_texts(self.db_path), ["restored"])
+        self.assertIsNot(self.api.db, old_db)  # the restored library adopted
+        self.assertEqual(len(self.backups()), 1)  # the safety backup is kept
+        self.assertEqual(self.stages(), [])
+        self.assertEqual(self.uploads(), [])
+        self.assertIsNone(self.api._jobs.active(RESTORE_JOB_KIND))
+
+    async def test_cancel_during_staging_waits_and_changes_nothing(self):
+        self.save("old")
+        gate = self.gate(PromptModel.stage_restore)
+        with (
+            self.fake_upload(["restored"]),
+            patch.object(PromptModel, "stage_restore", gate.method()),
+        ):
+            task = asyncio.ensure_future(self.api.restore_database(None))
+            await self.wait_thread(gate.entered)
+            task.cancel()
+            await self.assert_still_pending(task)
+            self.assertEqual(len(self.uploads()), 1)  # still being read
+            self.assertIsNotNone(self.api._jobs.active(RESTORE_JOB_KIND))
+            gate.release.set()
+            with self.assertRaises(asyncio.CancelledError):
+                await asyncio.wait_for(task, WAIT)
+        self.assertEqual(_texts(self.db_path), ["old"])
+        self.assertEqual(self.backups(), [])
+        self.assertEqual(self.stages(), [])  # the finished stage is removed
+        self.assertEqual(self.uploads(), [])
+        self.assertIsNone(self.api._jobs.active(RESTORE_JOB_KIND))
+
+    async def test_a_publish_that_fails_after_the_cancel_still_cancels(self):
+        self.save("old")
+
+        def refuse(model, staged):
+            raise sqlite3.OperationalError("refused")
+
+        gate = self.gate(refuse)
+        with (
+            self.fake_upload(["restored"]),
+            patch.object(PromptModel, "publish_staged", gate.method()),
+        ):
+            task = asyncio.ensure_future(self.api.restore_database(None))
+            await self.wait_thread(gate.entered)
+            task.cancel()
+            await self.assert_still_pending(task)
+            gate.release.set()
+            # Not answered as a 500: the cancel goes on.
+            with self.assertRaises(asyncio.CancelledError):
+                await asyncio.wait_for(task, WAIT)
+        self.assertEqual(_texts(self.db_path), ["old"])
+        self.assertEqual(self.stages(), [])
+        self.assertEqual(self.uploads(), [])
+        self.assertIsNone(self.api._jobs.active(RESTORE_JOB_KIND))
+
+    async def test_a_second_cancel_still_adopts_the_restored_library(self):
+        self.save("old")
+        old_db = self.api.db
+        publish = self.gate(PromptModel.publish_staged)
+        prune = self.gate(admin_mod._prune_safety_backups)
+        with (
+            self.fake_upload(["restored"]),
+            patch.object(PromptModel, "publish_staged", publish.method()),
+            patch.object(admin_mod, "_prune_safety_backups", prune),
+        ):
+            task = asyncio.ensure_future(self.api.restore_database(None))
+            await self.wait_thread(publish.entered)
+            task.cancel()
+            await self.assert_still_pending(task)
+            publish.release.set()
+            await self.wait_thread(prune.entered)  # adopting the library
+            task.cancel()  # the loop's own shutdown cancels again
+            await self.assert_still_pending(task)
+            prune.release.set()
+            with self.assertRaises(asyncio.CancelledError):
+                await asyncio.wait_for(task, WAIT)
+        self.assertEqual(_texts(self.db_path), ["restored"])
+        self.assertIsNot(self.api.db, old_db)
+        self.assertEqual(len(self.backups()), 1)
+        self.assertEqual(self.stages(), [])
+        self.assertIsNone(self.api._jobs.active(RESTORE_JOB_KIND))
+
+
+class TestReviewedRestoreAtShutdown(ShutdownRestoreCase):
+    async def test_cancelled_job_waits_for_the_publish(self):
+        self.save("old")
+        old_db = self.api.db
+        body = await self.preview(self.upload(["restored"]))
+        gate = self.gate(PromptModel.publish_staged)
+        with patch.object(PromptModel, "publish_staged", gate.method()):
+            started = await self.apply(body["token"])
+            job = self.api._jobs.get(started["job"]["id"])
+            await self.wait_thread(gate.entered)
+            job.task.cancel()
+            await self.assert_still_pending(job.task)
+            self.assertEqual(len(self.stages()), 1)
+            self.assertTrue(job.running)  # its exclusive admission holds
+            gate.release.set()
+            with self.assertRaises(asyncio.CancelledError):
+                await asyncio.wait_for(job.task, WAIT)
+        self.assertEqual(_texts(self.db_path), ["restored"])
+        self.assertIsNot(self.api.db, old_db)
+        self.assertEqual(len(self.backups()), 1)
+        self.assertEqual(self.stages(), [])
+        self.assertFalse(job.running)

tests/test_sampler_less_prompt.py

@@ -0,0 +1,112 @@
+"""``sampler_less_prompt``: the one wired text of a graph with no sampler (#252)."""
+
+import time
+import unittest
+
+from utils.prompt_pair_resolver import sampler_less_prompt
+
+
+def _guided(**nodes):
+    """``nodes`` plus a guider consuming node ``"1"``."""
+    graph = {"9": {"class_type": "BasicGuider", "inputs": {"conditioning": ["1", 0]}}}
+    graph.update(nodes)
+    return graph
+
+
+class TestSamplerLessPrompt(unittest.TestCase):
+    def test_not_a_graph_or_too_large_is_absent(self):
+        for graph in (None, [], {}, "text"):
+            self.assertEqual(sampler_less_prompt(graph)["state"], "absent")
+        big = {str(i): {"class_type": "X", "inputs": {}} for i in range(3)}
+        self.assertEqual(sampler_less_prompt(big, max_visits=2)["state"], "absent")
+
+    def test_one_literal_text(self):
+        graph = _guided(**{"1": {"class_type": "Enc", "inputs": {"text": " a fox "}}})
+        self.assertEqual(
+            sampler_less_prompt(graph),
+            {"state": "resolved", "text": "a fox", "texts": ["a fox"]},
+        )
+
+    def test_a_number_literal_is_never_a_prompt(self):
+        for value in (2048, "2048", "1.5", " 7 "):
+            graph = _guided(**{"1": {"class_type": "Enc", "inputs": {"text": value}}})
+            self.assertEqual(sampler_less_prompt(graph)["state"], "absent", value)
+
+    def test_a_long_digit_run_with_a_letter_is_text_and_fast(self):
+        text = "9" * 50_000 + "x"
+        graph = _guided(**{"1": {"class_type": "Enc", "inputs": {"text": text}}})
+        started = time.monotonic()
+        self.assertEqual(sampler_less_prompt(graph)["text"], text)
+        self.assertLess(time.monotonic() - started, 1.0)
+
+    def test_number_like_words_are_prompts(self):
+        for value in ("nan", "inf", "Infinity", "1_000"):
+            graph = _guided(**{"1": {"class_type": "Enc", "inputs": {"text": value}}})
+            self.assertEqual(sampler_less_prompt(graph)["text"], value)
+
+    def test_an_own_node_with_unreadable_text_beside_another_text_is_ambiguous(self):
+        graph = _guided(
+            **{
+                "1": {"class_type": "PromptManager", "inputs": {"text": ["5", 0]}},
+                "5": {"class_type": "LLM", "inputs": {"seed": 3}},
+                "2": {"class_type": "Enc", "inputs": {"text": "<lora:x:1>"}},
+                "3": {"class_type": "Sink", "inputs": {"c": ["2", 0]}},
+            }
+        )
+        self.assertEqual(sampler_less_prompt(graph)["state"], "ambiguous")
+
+    def test_unhashable_class_type_is_an_unknown_node_not_a_crash(self):
+        graph = _guided(**{"1": {"class_type": ["x"], "inputs": {"text": "a fox"}}})
+        self.assertEqual(sampler_less_prompt(graph)["text"], "a fox")
+
+    def test_own_negative_node_is_not_authoritative(self):
+        graph = _guided(
+            **{
+                "1": {"class_type": "Enc", "inputs": {"text": "a fox"}},
+                "2": {
+                    "class_type": "PromptManagerNegative",
+                    "inputs": {"text": "blur"},
+                },
+                "3": {"class_type": "Sink", "inputs": {"c": ["2", 0]}},
+            }
+        )
+        found = sampler_less_prompt(graph)
+        self.assertEqual(found["state"], "ambiguous")
+        self.assertEqual(found["texts"], ["a fox", "blur"])
+
+    def test_own_node_text_wins_over_an_untraceable_text(self):
+        graph = _guided(
+            **{
+                "1": {"class_type": "PromptManagerText", "inputs": {"text": "a fox"}},
+                "2": {"class_type": "Enc", "inputs": {"text": ["5", 0]}},
+                "5": {"class_type": "LLM", "inputs": {"seed": 3}},
+                "6": {"class_type": "Sink", "inputs": {"c": ["2", 0]}},
+            }
+        )
+        self.assertEqual(sampler_less_prompt(graph)["text"], "a fox")
+
+    def test_two_own_texts_are_ambiguous(self):
+        graph = _guided(
+            **{
+                "1": {"class_type": "PromptManager", "inputs": {"text": "a fox"}},
+                "2": {"class_type": "PromptManager", "inputs": {"text": "a cat"}},
+                "3": {"class_type": "Sink", "inputs": {"c": ["2", 0]}},
+            }
+        )
+        self.assertEqual(sampler_less_prompt(graph)["state"], "ambiguous")
+
+    def test_only_an_untraceable_text_is_ambiguous_with_nothing_to_show(self):
+        graph = _guided(
+            **{
+                "1": {"class_type": "Enc", "inputs": {"text": ["5", 0]}},
+                "5": {"class_type": "LLM", "inputs": {"seed": 3}},
+            }
+        )
+        self.assertEqual(
+            sampler_less_prompt(graph),
+            {"state": "ambiguous", "text": None, "texts": []},
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_thumbnail_auto.py

@@ -978,6 +978,27 @@ class TestThumbnailOnCapture(AutoThumbnailCase):
                 (self.output_dir / "thumbnails" / f"{name}_thumb.webp").is_file()
             )
 
+    async def test_a_worker_refused_at_interpreter_exit_skips_quietly(self):
+        """#375: refused once the interpreter exits; nothing stays queued."""
+
+        class Refused(threading.Thread):
+            def start(self):
+                raise RuntimeError("can't create new thread at interpreter shutdown")
+
+        namespace = type(sys)("threading_copy")
+        namespace.__dict__.update(vars(threading))
+        namespace.Thread = Refused
+        thumbnailer = self.api._auto_thumbnails()
+        path = make_png(self.output_dir / "late.png", size=(32, 32))
+        with (
+            patch.object(auto, "threading", namespace),
+            self.assertNoLogs("prompt_manager.api", "INFO"),
+        ):
+            self.assertEqual(self.api.queue_capture_thumbnail(str(path)), "skipped")
+        self.assertFalse(thumbnailer._worker_running)
+        self.assertEqual(len(thumbnailer._queue), 0)
+        self.assertTrue(thumbnailer.wait_idle(0))
+
     def test_queue_is_bounded(self):
         thumbnailer = auto.AutoThumbnailer(
             lambda *a: {"action": "generated"},

utils/image_monitor.py

@@ -35,6 +35,7 @@ from watchdog.events import FileSystemEventHandler
 
 from .hashing import generate_prompt_hash
 from .metadata_extractor import ComfyUIMetadataExtractor
+from .parallel import note_refusal, refused_at_shutdown
 from .logging_config import get_logger
 from .prompt_graph import resolve_text, run_prompt_nodes
 from .prompt_pair_resolver import resolve_prompt_pair
@@ -194,8 +195,9 @@ class ImageGenerationHandler(FileSystemEventHandler):
         and once its size has stopped changing.
 
         Returns:
-            False when the path is already pending, the pending set is full or
-            the handler is closed
+            False when the path is already pending, the pending set is full,
+            the handler is closed or the interpreter is exiting (its worker
+            cannot start)
         """
         with self._pending_lock:
             if self._closed:
@@ -211,7 +213,15 @@ class ImageGenerationHandler(FileSystemEventHandler):
                 return False
             due = time.monotonic() + self.processing_delay
             self._pending[image_path] = (prompt_snapshot, due)
-            self._ensure_worker()
+            try:
+                self._ensure_worker()
+            except RuntimeError as error:
+                # No worker will ever take it: never leave it pending (#375).
+                self._pending.pop(image_path, None)
+                if not refused_at_shutdown(error):
+                    raise
+                note_refusal(self.logger, "The image worker")
+                return False
             self._work.put(image_path)
         return True
 

utils/parallel.py

@@ -5,17 +5,51 @@ a CUDA-initialised, multi-threaded process is unsafe and the spawn start
 method would re-import ComfyUI's main module in every child. Pillow releases
 the GIL while decoding, resizing and encoding, so threads still scale close
 to linearly for both jobs.
+
+Interpreter exit (#375): ``threading._shutdown`` shuts every thread pool
+(``submit`` then raises "cannot schedule new futures after interpreter
+shutdown") before it joins the non-daemon threads and long before the package
+teardown runs, while our daemon threads keep working. Measured on 3.12 and
+3.14, ``Thread.start`` still succeeds in that window; it is refused ("can't
+create new thread at interpreter shutdown") only once the interpreter is
+marked finalizing, after the ``atexit`` callbacks. ``refused_at_shutdown``
+recognises both; ``note_refusal`` logs one DEBUG line per place, with no
+traceback.
 """
 
+import logging
+import threading
 from concurrent.futures import ThreadPoolExecutor
 
+SHUTDOWN_REFUSAL = "interpreter shutdown"
+_logger = logging.getLogger("prompt_manager.parallel")
+_noted = set()
+_noted_lock = threading.Lock()
+
+
+def refused_at_shutdown(error):
+    """Whether ``error`` is a thread or pool start refused at interpreter exit."""
+    return isinstance(error, RuntimeError) and SHUTDOWN_REFUSAL in str(error)
+
+
+def note_refusal(logger, what):
+    """Log once per ``what``, at DEBUG and without a traceback, that ``what``
+    was refused because the interpreter is exiting."""
+    key = (logger.name, what)
+    with _noted_lock:
+        if key in _noted:
+            return
+        _noted.add(key)
+    logger.debug(f"{what}: refused, the interpreter is exiting")
+
 
 def map_parallel(func, items, workers):
     """Apply ``func`` to every item on up to ``workers`` threads, keeping order.
 
     ``workers`` of 1 (or fewer) runs on the calling thread with no pool. An
     exception in ``func`` propagates to the caller, so callers that must
-    survive bad input wrap ``func`` themselves.
+    survive bad input wrap ``func`` themselves. A pool refused at interpreter
+    exit runs the items it did not start on the calling thread (#375).
     """
     items = list(items)
     workers = max(1, int(workers or 1))
@@ -23,4 +57,35 @@ def map_parallel(func, items, workers):
         return [func(item) for item in items]
     max_workers = min(workers, len(items))
     with ThreadPoolExecutor(max_workers, thread_name_prefix="pm-worker") as pool:
-        return list(pool.map(func, items))
+        futures = []
+        try:
+            for item in items:
+                futures.append(pool.submit(func, item))
+        except RuntimeError as error:
+            if not refused_at_shutdown(error):
+                raise
+            note_refusal(_logger, "A worker pool")
+            # Drop what was queued but not started; it runs here instead.
+            # (A refused Thread.start leaves its item queued without a future
+            # here: it is cancelled too, unless a busy pool thread took it in
+            # between, the one case where an item can run twice.)
+            pool.shutdown(wait=True, cancel_futures=True)
+        else:
+            return [future.result() for future in futures]
+    return _run_rest(func, items, futures)
+
+
+def _run_rest(func, items, futures):
+    """The pool's finished results, and every item it did not run, run here.
+
+    After ``shutdown(cancel_futures=True)`` each submitted future either ran
+    (its result or exception is kept) or was cancelled before it started.
+    """
+    results = []
+    for index, item in enumerate(items):
+        future = futures[index] if index < len(futures) else None
+        if future is not None and not future.cancelled():
+            results.append(future.result())
+        else:
+            results.append(func(item))
+    return results

utils/prompt_pair_resolver.py

@@ -2578,3 +2578,115 @@ def resolve_text_input(
         return resolver.text(inputs.get(input_name), "string")[0]
     except _Unresolved:
         return None
+
+
+_NUMBER = re.compile(r"[+-]?(?:\d+(?:\.\d*)?|\.\d+)(?:[eE][+-]?\d+)?")
+
+
+def _numeric(text: str) -> bool:
+    """Whether ``text`` is a plain number (a widget value such as ``2048``)."""
+    return _NUMBER.fullmatch(text) is not None
+
+
+def sampler_less_prompt(graph: Any, *, max_visits: int = 500) -> Dict[str, Any]:
+    """The positive prompt of an API graph with no sampler pair, if one (#252).
+
+    For graphs ``_no_sampler`` finds ``absent`` (no node takes a ``positive``
+    or ``negative`` input, such as a Flux guider or a video wrapper). Only the
+    executed API graph is read, never the UI workflow. A candidate is a node
+    with a ``text`` input whose output another node consumes: a text that
+    feeds nothing did not make the image. Its text is traced as a STRING
+    (``resolve_text_input``); a linked text no rule reads is unknown unless
+    it is built from a candidate's text (``_covered``). Blank and
+    numeric texts are never prompts, and a node with a seed input holds a
+    template, not the prompt (unknown). This repository's own positive-text
+    nodes are authoritative: when they give a text, only they count.
+
+    Returns ``{"state", "text", "texts"}``: ``resolved`` with the one distinct
+    text; ``ambiguous`` when several texts or an unknown one could be the
+    prompt (``texts`` lists the readable ones); ``absent`` when none.
+    """
+    absent = {"state": "absent", "text": None, "texts": []}
+    if not isinstance(graph, dict) or not graph or len(graph) > max_visits:
+        return absent
+    nodes = {str(key): value for key, value in graph.items()}
+    consumed = {
+        str(value[0])
+        for node in nodes.values()
+        if isinstance(node, dict) and isinstance(node.get("inputs"), dict)
+        for value in node["inputs"].values()
+        if _link(value)
+    }
+    own: list = []
+    other: list = []
+    unknown = False
+    for node_id, node in nodes.items():
+        inputs = node.get("inputs") if isinstance(node, dict) else None
+        if node_id not in consumed or not _has_text_input(node):
+            continue
+        if _seeded(inputs):
+            # Generated at run time (a wildcard or a language model): its
+            # text input is a template, not the prompt that ran.
+            unknown = True
+            continue
+        value = inputs["text"]
+        text = resolve_text_input(nodes, node_id, "text", max_visits=max_visits)
+        if text is None:
+            # A literal that is not a string (a number) is never a prompt.
+            if _link(value) and not _covered(nodes, value):
+                unknown = True
+            continue
+        text = text.strip()
+        if not text or _numeric(text):
+            continue
+        kind = node.get("class_type")
+        if isinstance(kind, str) and kind in _OWN_CLASSES and "Negative" not in kind:
+            own.append(text)
+        else:
+            other.append(text)
+    texts = list(dict.fromkeys(own or other))
+    if len(texts) == 1 and (own or not unknown):
+        return {"state": "resolved", "text": texts[0], "texts": texts}
+    if texts or unknown:
+        return {"state": "ambiguous", "text": None, "texts": texts}
+    return absent
+
+
+def _has_text_input(node: Any) -> bool:
+    inputs = node.get("inputs") if isinstance(node, dict) else None
+    return isinstance(inputs, dict) and "text" in inputs
+
+
+def _seeded(inputs: dict) -> bool:
+    """A seed input: the node's string output is a run-time value (the
+    resolver's ``runtime-value`` rule)."""
+    return any(isinstance(name, str) and name.endswith("seed") for name in inputs)
+
+
+def _covered(nodes: dict, link: list) -> bool:
+    """Whether an unreadable text link is built from a candidate's text.
+
+    True when a node upstream of ``link`` is read through its first output
+    (slot 0, the legacy encoder rule's text) and has an unseeded ``text``
+    input: that candidate is read on its own, and whatever else is joined to
+    it at run time (a caption, a language model) is no stored prompt to
+    choose. Another output of such a node (a loader's trigger words) is not
+    its text. False when the text comes only from inputs no candidate holds
+    (a builder's ``prompt`` field): the prompt is unknown then.
+    """
+    seen, stack = set(), [(str(link[0]), link[1])]
+    while stack:
+        node_id, slot = stack.pop()
+        if (node_id, slot) in seen:
+            continue
+        seen.add((node_id, slot))
+        node = nodes.get(node_id)
+        inputs = node.get("inputs") if isinstance(node, dict) else None
+        if not isinstance(inputs, dict):
+            continue
+        if "text" in inputs and slot == 0 and not _seeded(inputs):
+            return True
+        stack.extend(
+            (str(value[0]), value[1]) for value in inputs.values() if _link(value)
+        )
+    return False