Merge remote-tracking branch 'origin/main' into fix/391-browser-flakes

* ci: pin black and flake8, let the bandit hook fail, cache npm in the JS job (#352)

The lint job installs black==26.10.1 and flake8==7.4.1 from
requirements-dev.txt (the versions CI resolved on main today), so a new
release cannot turn CI red with no code change.

The local bandit hook ran 'bash -c ... "$@"; true': it could never
fail, and bash bound the first staged file to $0 so it was never
scanned. It now runs 'bandit -lll -q' directly, failing on high
severity exactly like code-quality.yml.

The frontend modules job's setup-node caches npm keyed on
tools/vendor/package-lock.json.

* test(history): wait on the writer's state, not short wall-clock pauses (#352)

Four 0.08 s pauses before asserting that the worker had not probed yet
now wait until the idle worker has run a whole loop pass at the current
manual clock value (three clock reads), so a slow runner cannot end the
pause before the worker decides. A mutation that probes 20 ms early now
fails two tests deterministically.

The real-clock bounded close no longer asserts close(0.02) returned in
under 0.2 s wall time; it asserts the in-flight write is still blocked
and the worker alive, like the output writer test. The manual-clock
bound stays exact. The last 0.1 s non-blocking submit check uses the
1 s NON_BLOCKING_CEILING from #351.

* test(api): give each API fixture database its own temp folder (#346)

Every PromptManagerAPI sweeps its database's folder at startup for the
stages, uploads and safety-backup copies an interrupted restore left.
The API test fixtures created their database directly in the system
temp folder, so any other test process starting an API there (another
worktree's suite, a pre-push hook, an agent's targeted run) deleted
this process's in-flight restore files. The restore, epoch and
retirement tests then failed with 'staged restore changed', 'unable to
open database file' or 503 instead of 409. Busy machines fail because
they are running other suites, not because they are slow: the same
CPU load with a private TMPDIR passes.

private_temp_db() puts the fixture database in a new folder removed at
cleanup (with any safety backups a restore left beside it). A
regression test sweeps a stand-in shared folder in the middle of a
multipart restore, as another suite's startup would.

* test(history): count worker parks, hold saturation blocks until released (#352)

wait_for_worker_pass now counts the idle worker's parks on its wake
event (two parks from now = one whole pass begun after the call)
instead of assuming a fixed number of clock reads per pass, which a
future clock read would have made vacuous. The real-clock bounded
close keeps a loose upper bound (0.02 s + NON_BLOCKING_CEILING).

The two saturation tests (history and output writer) blocked the
worker with release.wait(3) while they queued 1023 more items. On a
loaded runner the fill took longer than 3 s, the worker unblocked on
its own and drained part of the queue (pending 868..1017 != 1024),
and cleanup then reported a leaked worker. The block now holds until
the test releases it; 60 s is only a hang guard, and cleanup always
sets the release. Under one core with 8 busy loops: 8/8 runs failed
before, 6/6 pass after.

* test(maintenance): wait on duplicate jobs with a disk-bound hang guard (#409)

On Windows py3.13 CI (PR #405) a one-file duplicate cleanup missed its
10 s WAIT. The test already waits on the job's own task; the 10 s bound
acted as a deadline on disk-bound work. In the same 50 s window the
following tests' setups took up to 9 s against 0.1-0.6 s around it
while the test bodies stayed fast, which is consistent with a
process-wide disk I/O stall on the runner rather than a lease, handle,
executor or job stall (the log has no thread dump to show more).

tests/disk_waits.py holds the one disk-bound hang guard (90 s per wait,
under the 120 s per-test watchdog), shared with RESTORE_SECONDS (#382).
Waits on find and cleanup job tasks in the duplicate-files tests use
it; gates and server-state polls keep WAIT. The keeper-validation test
turns the safety backup off, since it does not test backups.

.github/workflows/test.yml

@@ -221,7 +221,10 @@ jobs:
       - uses: actions/setup-python@v7
         with:
           python-version: "3.12"
-      - run: pip install black flake8
+          cache: pip
+          cache-dependency-path: requirements-dev.txt
+      # The exact black and flake8 pins from requirements-dev.txt, so a new release cannot turn CI red.
+      - run: pip install $(grep -E '^(black|flake8)==' requirements-dev.txt)
       - run: black --check .
       - run: flake8 .
 
@@ -235,6 +238,8 @@ jobs:
       - uses: actions/setup-node@v7
         with:
           node-version: "24"
+          cache: npm
+          cache-dependency-path: tools/vendor/package-lock.json
 
       - name: Check shipped extension and page modules parse
         run: |

.pre-commit-config.yaml

@@ -13,9 +13,11 @@ repos:
         language: system
         types: [python]
 
+      # Fails at the severity CI enforces (code-quality.yml: -lll, high only), never stricter.
+      # pre-commit appends the staged files, so no shell wrapper can drop the first one.
       - id: bandit
         name: bandit
-        entry: bash -c 'bandit "$@" -ll -q 2>&1; true'
+        entry: bandit -lll -q
         language: system
         types: [python]
         exclude: ^(tests/|reference/)

TESTING.md

@@ -3200,12 +3200,16 @@ the per-section hand-off panel, was retired by S12 (#329): every section is nati
   again"; an Unload that found nothing loaded only reads the status again. At phone width
   the concept's 120px select cap cuts every engine label, so `#tag-model` alone keeps its
   natural width there and its row wraps the select under the text (`workspace.css`
-  `/* D132 */`); Existing tags stays as the concept draws it, cut at 390 as there. The
-  comparison maps `selectsFit` only for the engine select and measures each row's
-  `flexWrap` and whether its control sits under the text. Pinned by `auto-tagging matches
-  the concept` (both themes and sizes), `auto-tagging at 390 wraps only the engine row`,
-  `auto-tagging engine select fits at 320 wide` and the mutation check (it includes a
-  wrapped row) (system-tagging-concept.spec), and the model, Download, Unload and Preview
+  `/* D132 */`). The concept's cap cuts Existing tags too ("Replace after review"); by the
+  maintainer's decision (#398) that parity is not kept: `#tag-existing` never shrinks for
+  its row text and at phone width keeps its natural width beside the text (its row does
+  not wrap; the text column narrows). 1600 is as the concept draws it. The comparison
+  expects `selectsFit` true for every select and measures each row's `flexWrap` and
+  whether its control sits under the text. Pinned by `auto-tagging matches the concept`
+  (both themes and sizes), `auto-tagging at 390 wraps only the engine row`, `auto-tagging
+  selects fit every option at 390/320 wide` (every option of both selects, both themes, no
+  page scroll) and the mutation check (a wrapped row, a squeezed engine and Existing tags
+  select) (system-tagging-concept.spec), and the model, Download, Unload and Preview
   tests of workspace-system-tagging.spec.
 - **D133** (Auto-tagging, S11): a "Suggestions per image" row (`#tag-suggestions`,
   number 1–100, host key `autotag_suggestion_limit`) follows the two confidence rows:

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"

requirements-dev.txt

@@ -8,8 +8,8 @@ coverage>=7.16.2
 numpy>=2.2.6  # WD14 engine tests
 
 # Code quality
-black>=26.10.1
-flake8>=7.4.1
+black==26.10.1
+flake8==7.4.1
 
 # Security scanning
 bandit[toml]>=1.9.4

tests/browser/images-concept.spec.mjs

@@ -286,7 +286,8 @@ const continuationText=page=>page.locator('.infinite-continuation > p').first().
 async function searchOf(page,term) {
     const before=await countText(page);
     await page.locator('#search').fill(term);
-    await expect.poll(()=>countText(page)).not.toBe(before);
+    // Production's caption reads "Loading images…" until the search's listing answers (#391); the concept's is instant.
+    await expect.poll(async()=>{const text=await countText(page);return text!==before&&text!=='Loading images…';}).toBe(true);
     await expect.poll(()=>page.locator(CARDS).count()).toBeLessThan(FIXTURE_IMAGES);
     await settle(page);
     const result={count:await countText(page),continuation:await continuationText(page),cards:await page.locator(`${CARDS} .image-open`).evaluateAll(nodes=>nodes.map(node=>node.getAttribute('aria-label')))};

tests/browser/inspector-concept.spec.mjs

@@ -228,7 +228,8 @@ async function productionSendOf(page,context) {
 async function tagOf(page) {
     const before=await page.locator('#result-count').textContent();
     await panelOf(page).locator('.detail-inner > .tags').last().getByRole('button',{name:'cinematic',exact:true}).click();
-    await expect.poll(()=>page.locator('#result-count').textContent()).not.toBe(before);
+    // Production's caption reads "Loading images…" until the tag's listing answers (#391); the concept's search is instant.
+    await expect.poll(async()=>{const text=await page.locator('#result-count').textContent();return text!==before&&text!=='Loading images…';}).toBe(true);
     await settle(page);
     return {count:await page.locator('#result-count').textContent(),
         cards:await page.locator(`${CARDS} .image-open`).evaluateAll(nodes=>nodes.map(node=>node.getAttribute('aria-label')))};
@@ -328,6 +329,25 @@ test('a recipe-linked image without recipe metadata shows its positive prompt me
     await expect(panelOf(page).locator('.detail-footer button')).toHaveText(['Send to ComfyUI','Prompt history']);
 });
 
+// #391: a tag chip replaces the collection; until its listing answers the caption says it is loading, never that the
+// filter matched nothing ("0 of 0 images" was painted before the response and read by a fast reader).
+test('a tag chip collection reads as loading, not empty, until its listing answers (#391)',async({page})=>{
+    await page.setViewportSize({width:1600,height:1100});
+    await openProductionPanel(page,'dark');
+    let release,arrived;
+    const held=new Promise(resolve=>{release=resolve;}),reached=new Promise(resolve=>{arrived=resolve;});
+    await page.route(url=>url.pathname==='/prompt_manager/images/output'&&url.searchParams.get('tags')==='cinematic',
+        async route=>{arrived();await held;await route.continue();},{times:1});
+    const count=page.locator('#result-count');
+    await panelOf(page).locator('.detail-inner > .tags').last().getByRole('button',{name:'cinematic',exact:true}).click();
+    await reached;
+    await expect(count).toHaveText('Loading images…');
+    await expect(page.locator(CARDS)).toHaveCount(0);
+    release();
+    await expect(count).toHaveText(countCopy('8 of 8 images · 2 positive prompts'));
+    await expect(page.locator(CARDS)).toHaveCount(8);
+});
+
 // Task C3 (D88, #249 PR C): the concept's chip row is the image's own tags (`i.tags`, app.js a402). With those names
 // saved as the file's current image tags, production draws the row from them (Tags filter buttons in the panel, plain
 // in the viewer, the linked positive's equal tags not repeated) and still equals the concept, in both sizes.

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

@@ -127,9 +127,9 @@ function mapTaggingConcept(reference, { template, phone }) {
         actions: { ...reference.actions, buttons: [preview, ['Download again', ''], ['Unload', '']] },
         // D134: the prompt-tag batch, reviewed first.
         operations: [{ tree: ['div', 'div.row-actions'], title: 'Tag output images', description: BATCH, actions: [['Review', '']] }],
-        // D132: the engine select shows its whole chosen option at every width (at 390 the concept's 120px cap cuts it;
-        // `#tag-model` alone may grow there). Existing tags keeps the concept's own fit, cut or not.
-        selectsFit: reference.selectsFit.map((fit, index) => (index === 0 ? true : fit)),
+        // D132 (#398): every select shows its whole chosen option at every width (at 390 the concept's 120px cap cuts
+        // them; production lets each grow past it).
+        selectsFit: reference.selectsFit.map(() => true),
     };
 }
 
@@ -168,16 +168,28 @@ for (const theme of THEMES) for (const [width, height] of SIZES) {
     });
 }
 
-// D132 below the concept's smallest size: on a 320px phone the engine select still shows its whole option.
-for (const theme of THEMES) test(`auto-tagging engine select fits at 320 wide ${theme}`, async ({ page }) => {
-    await page.setViewportSize({ width: 320, height: 700 });
-    const production = await productionTagging(page, theme);
-    for (const model of ['hf', 'gguf', 'wd14-swinv2']) {
-        await page.locator('#workspace-main #tag-model').selectOption(model);
-        expect((await page.evaluate(measureTagging, '#workspace-main')).selectsFit[0], model).toBe(true);
-    }
-    expect(production.pageScroll).toBe(false);
-});
+// D132 (#398) at phone width and below the concept's smallest size: every option of every select, once chosen, shows
+// whole, with no page scroll.
+for (const theme of THEMES) for (const width of [390, 320]) {
+    test(`auto-tagging selects fit every option at ${width} wide ${theme}`, async ({ page }, testInfo) => {
+        await page.setViewportSize({ width, height: 844 });
+        await productionTagging(page, theme);
+        await page.locator('#workspace-main #tag-existing').selectOption('replace_after_review');
+        await page.locator('#workspace-main section.surface').screenshot({ path: testInfo.outputPath(`production-tagging-selects-${theme}-${width}.png`) });
+        const selects = page.locator('#workspace-main section.surface > .setting-row > select');
+        await expect(selects).toHaveCount(2);
+        for (let index = 0; index < 2; index += 1) {
+            const select = selects.nth(index);
+            const values = await select.evaluate(node => [...node.options].map(option => option.value));
+            for (const value of values) {
+                await select.selectOption(value);
+                const measured = await page.evaluate(measureTagging, '#workspace-main');
+                expect(measured.selectsFit, `${await select.getAttribute('id')}=${value}`).toEqual([true, true]);
+                expect(measured.pageScroll, value).toBe(false);
+            }
+        }
+    });
+}
 
 test('auto-tagging comparison detects mutations', async ({ page, request }) => {
     await page.setViewportSize({ width: 1600, height: 1100 });
@@ -222,6 +234,9 @@ test('auto-tagging comparison detects mutations', async ({ page, request }) => {
     const undoSqueeze = await mutate(page, { css: '#workspace-main #tag-model { max-width: 60px !important; min-width: 0 !important; }' });
     expect(await paths()).toEqual(expect.arrayContaining(['selectsFit.0']));
     await undoSqueeze();
+    const undoSqueezeExisting = await mutate(page, { css: '#workspace-main #tag-existing { max-width: 60px !important; min-width: 0 !important; }' });
+    expect(await paths()).toEqual(expect.arrayContaining(['selectsFit.1']));
+    await undoSqueezeExisting();
     // A row that wraps where the concept's does not.
     const undoWrap = await mutate(page, { css: '#workspace-main .setting-row:has(> #tag-existing) { flex-wrap: wrap; }' });
     expect(await paths()).toEqual(['rows.1.flexWrap']);
@@ -233,14 +248,14 @@ test('auto-tagging comparison detects mutations', async ({ page, request }) => {
     expect(diffInventory(expected, await measure())).toEqual([]);
 });
 
-// D132 at phone width, unmapped: only the engine row wraps; Existing tags keeps the concept's cut select.
+// D132 at phone width: only the engine row wraps; Existing tags grows beside its text instead (#398).
 test('auto-tagging at 390 wraps only the engine row', async ({ page, request }) => {
     await page.setViewportSize({ width: 390, height: 844 });
     const concept = await conceptTagging(page, 'dark');
     const production = await productionTagging(page, 'dark');
     expect(diffInventory(mapTaggingConcept(concept, await savedTemplate(request, page)), production)).toEqual([]);
     expect(production.rows.map(row => row.flexWrap)).toEqual(['wrap', 'nowrap', 'nowrap', 'nowrap', 'nowrap']);
-    expect(production.selectsFit[1]).toBe(concept.selectsFit[1]);
+    expect(production.selectsFit).toEqual([true, true]);
 });
 
 /** Runs in the page: the open review dialog: its title, body and the concept's tokens. */

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

@@ -47,6 +47,17 @@ async function serverJson(request, url) {
     return response.json();
 }
 
+/**
+ * The restore job's own end (#361): its progress stream closes when the job takes its final state, however long the
+ * disk-bound publish takes (bounded by the test, not by a 5 s expect), and the job finished. The toast the page shows
+ * for it is then the page's reaction alone, checked with the default timeout.
+ */
+async function restoreFinished(request, id) {
+    await expect(await request.get(`/prompt_manager/jobs/${id}/events`)).toBeOK();
+    const { job } = await serverJson(request, `/prompt_manager/jobs/${id}`);
+    expect([job.kind, job.state]).toEqual(['restore', 'finished']);
+}
+
 async function openBackups(page) {
     // A fresh page each time: the drafts live for the page session (Q10), and a hash-only goto would not reload.
     if (page.url().endsWith('#/system/backups')) await page.reload();
@@ -216,6 +227,7 @@ test('Restore starts the job, opens Activity; views reload after it finishes', a
     const { job } = await (await applied).json();
     expect([job.kind, job.title, job.cancellable]).toEqual(['restore', 'Database restore', false]);
     await expect(page).toHaveURL(`${workspace}#/activity/${job.id}`);
+    await restoreFinished(request, job.id);
     await expect(toast(page)).toHaveText(/^Database restore finished · Restored \d+ prompts?, /);
     // The library is the backup's again, and the page says so without a reload: the sidebar counts were read again.
     expect(await libraryTotal(request)).toBe(restoredTotal);
@@ -244,7 +256,8 @@ test('a finished restore shows the open page again from the restored data', asyn
     const card = page.locator('#workspace-main article.prompt-card').filter({ hasText: 'S9 prompt the restore removes' });
     await expect(card).toHaveCount(1);
     await expect(await request.post('/__test__/release')).toBeOK();
-    await applied;
+    const { job } = await (await applied).json();
+    await restoreFinished(request, job.id);
     await expect(toast(page)).toHaveText(/^Database restore finished · /);
     // The shelf was read again from the restored library, on the same route.
     await expect(card).toHaveCount(0);
@@ -422,6 +435,7 @@ test('a refused restore reopens its review: Restore retries it, Discard releases
     await dialog(page).getByRole('button', { name: 'Restore', exact: true }).click();
     const { job } = await (await applied).json();
     await expect(page).toHaveURL(`${workspace}#/activity/${job.id}`);
+    await restoreFinished(request, job.id);
     await expect(toast(page)).toHaveText(/^Database restore finished · /);
     expect(await stagesOf(request)).toBe(0);
 });

tests/browser/workspace_system_fixture.py

@@ -35,7 +35,8 @@ read. These per-server hooks only change what a System test needs on top:
   previews for Convert outdated;
 - ``POST /__test__/system/jxl-encoder`` ``{available}`` (#317) makes the
   server read as having (``cjxl``) or lacking a JPEG XL encoder without
-  probing; reset restores the thumbnail settings and the real detection.
+  probing, whatever the host has installed (#371: ``false`` stands in for the
+  detector itself); reset restores the thumbnail settings and the real detection.
 
 - ``POST /__test__/system/duplicate-prompts`` (S7) adds two prompts whose
   text differs only in case and spacing (a duplicate prompt text group);
@@ -118,6 +119,11 @@ def register_system_commands(app, routes, api, root: Path):
     }
     from py.api import thumbnail_encode
 
+    real_detect = thumbnail_encode._detect
+
+    def _no_encoder():
+        return None
+
     monitor = getattr(image_monitor, "_monitor_instance", None)
     failures: dict[str, int] = {}
 
@@ -147,6 +153,7 @@ def register_system_commands(app, routes, api, root: Path):
         api._health_report = None
         for (owner, name), value in host_values.items():
             setattr(owner, name, value)
+        thumbnail_encode._detect = real_detect
         thumbnail_encode.reset_encoders()
         image_monitor._monitor_instance = monitor
         api.__dict__.pop("_library_resolver", None)
@@ -298,7 +305,12 @@ def register_system_commands(app, routes, api, root: Path):
         available = data.get("available")
         if not isinstance(available, bool):
             raise web.HTTPBadRequest()
-        # Test-only: stand in the cached detection (no probe).
+        # Test-only: the fixture, not the host, decides (#371). Without an encoder the
+        # detector itself stands in, so a re-probe (a "none found" is probed again on
+        # every settings read) cannot find the host's real encoder; with one the cached
+        # answer stands, and a failed write still probes the real detector as before.
+        # Reset restores the real detector.
+        thumbnail_encode._detect = real_detect if available else _no_encoder
         with thumbnail_encode._detect_lock:
             thumbnail_encode._detected = "cjxl" if available else None
         return web.json_response({"success": True})

tests/disk_waits.py

@@ -0,0 +1,16 @@
+"""Hang guards for waits on disk-bound work (tests only).
+
+A test that waits for a job or handoff whose work is file and SQLite I/O
+(a safety backup with fsync, a restore, hashing and unlinking files) waits
+on that state; the ceiling below only ends a real hang. Hosted Windows
+runners stall disk I/O for tens of seconds at a time: a restore took over
+10 s (merge queue, #382), and on PR #405 every test setup in a 50 s window
+took up to 9 s against 0.1-0.6 s around it while the test bodies stayed
+fast, so a one-file duplicate cleanup missed a 10 s bound (#409).
+
+The ceiling is per wait and stays under the suite's 120 s per-test hang
+watchdog. A test with several such waits in a row can reach the watchdog
+first; either way a real hang fails, the watchdog with every thread's stack.
+"""
+
+DISK_BOUND_SECONDS = 90

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

@@ -234,6 +234,23 @@ test('the count line reads N of M images and K positive prompts only when known'
     assert.equal(countLine({loaded:1200,total:3400,positiveCount:0}),'1,200 of 3,400 images · 0 positive prompts');
 });
 
+test('the count line reads as loading, never as an empty result, until a first page answers (#391)', async () => {
+    const {countLine,createImageStream}=await source;
+    assert.equal(countLine({loaded:0,total:0,positiveCount:null,loading:true,error:null}),'Loading images…');
+    // A failed first page and a settled empty one say what the server said; a later page keeps the loaded count.
+    assert.equal(countLine({loaded:0,total:0,positiveCount:null,loading:false,error:new Error('x')}),'0 of 0 images');
+    assert.equal(countLine({loaded:0,total:0,positiveCount:0,loading:false,error:null}),'0 of 0 images · 0 positive prompts');
+    assert.equal(countLine({loaded:40,total:102,positiveCount:2,loading:true,error:null}),'40 of 102 images · 2 positive prompts');
+    // While a reset's first page is in flight the line reads as loading (the cleared state before it is replaced in
+    // the same task, so it is never painted).
+    let answer;const lines=[];
+    const stream=createImageStream({api:{get:()=>new Promise(resolve=>{answer=resolve;})},onChange:state=>lines.push(countLine(state))});
+    const loading=stream.reset({tags:['cinematic']});
+    assert.equal(lines.at(-1),'Loading images…');
+    answer({images:[{path:'a.png',root_index:0}],total:1,has_more:false,next_cursor:null,positive_count:1});await loading;
+    assert.equal(lines.at(-1),'1 of 1 image · 1 positive prompt');
+});
+
 test('the caption date names the newest result day under Newest first only', async () => {
     const {captionDate}=await source;
     const now=new Date(2026,9,17,15,0);

tests/private_temp.py

@@ -0,0 +1,24 @@
+"""A temp SQLite database in a folder no other test process shares (#346).
+
+``PromptManagerAPI`` sweeps its database's folder at startup for the stages,
+uploads and safety-backup copies an interrupted restore or backup left. A
+fixture database directly in the system temp folder made every other test
+process that started an API (another worktree's suite, a pre-push hook) delete
+this process's in-flight restore files, and left safety backups behind there.
+"""
+
+import tempfile
+
+
+def private_temp_db(case):
+    """Return a closed ``NamedTemporaryFile`` handle for an empty ``.db`` file.
+
+    The file lives in a new folder that ``case`` removes at cleanup, after its
+    own teardown has closed and unlinked the database (cleanups run after
+    ``tearDown``), together with anything a restore or backup left next to it.
+    """
+    folder = tempfile.TemporaryDirectory(ignore_cleanup_errors=True)
+    case.addCleanup(folder.cleanup)
+    handle = tempfile.NamedTemporaryFile(delete=False, suffix=".db", dir=folder.name)
+    handle.close()
+    return handle

tests/test_api.py

@@ -8,7 +8,6 @@ without requiring a full ComfyUI server.
 import json
 import os
 import sys
-import tempfile
 import unittest
 
 sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
@@ -18,6 +17,7 @@ from aiohttp.test_utils import AioHTTPTestCase
 
 from database.operations import PromptDatabase
 from py.api import PromptManagerAPI
+from tests.private_temp import private_temp_db
 from utils.hashing import generate_prompt_hash
 
 
@@ -25,8 +25,7 @@ class APITestCase(AioHTTPTestCase):
     """Base class that stands up a test aiohttp app with PromptManager routes."""
 
     async def get_application(self):
-        self._temp_db = tempfile.NamedTemporaryFile(delete=False, suffix=".db")
-        self._temp_db.close()
+        self._temp_db = private_temp_db(self)
 
         app = web.Application()
         routes = web.RouteTableDef()

tests/test_api_autotag.py

@@ -35,6 +35,7 @@ from aiohttp.test_utils import AioHTTPTestCase
 
 from database.operations import PromptDatabase
 from py.api import PromptManagerAPI
+from tests.private_temp import private_temp_db
 from py.api.autotag_routes import MODEL_BUSY_ERROR
 from py.api.jobs import JobBusy
 from utils.hashing import generate_prompt_hash
@@ -143,8 +144,7 @@ class AutotagAPITestCase(AioHTTPTestCase):
     fake engine."""
 
     async def get_application(self):
-        self._temp_db = tempfile.NamedTemporaryFile(delete=False, suffix=".db")
-        self._temp_db.close()
+        self._temp_db = private_temp_db(self)
         self.tmp_root = Path(tempfile.mkdtemp())
         self.output_dir = self.tmp_root / "output"
         self.output_dir.mkdir()

tests/test_api_duplicate_files.py

@@ -18,15 +18,19 @@ from pathlib import Path
 from unittest.mock import patch
 
 try:
+    from tests.disk_waits import DISK_BOUND_SECONDS
     from tests.open_handles import open_handles_under
     from tests.test_api_maintenance import ApplyTestCase, _digest
 except ImportError:  # discovered with tests/ as the top-level directory
+    from disk_waits import DISK_BOUND_SECONDS
     from open_handles import open_handles_under
     from test_api_maintenance import ApplyTestCase, _digest
 
 from py.api import duplicate_files  # noqa: E402
 from py.config import GalleryConfig, PromptManagerConfig  # noqa: E402
 
+# Gates and server-state polls. Waits on a find or cleanup job's task use
+# DISK_BOUND_SECONDS: those jobs hash, back up and unlink files (#409).
 WAIT = 10
 FIND = "/prompt_manager/jobs/duplicates"
 GROUPS = "/prompt_manager/maintenance/duplicates/"
@@ -83,7 +87,7 @@ class DuplicateFilesTestCase(ApplyTestCase):
         self.assertEqual(response.status, 202, text)
         self.assertNotIn(self.tmpdir, text)
         job = self.api._jobs.get(json.loads(text)["job"]["id"])
-        await asyncio.wait_for(asyncio.shield(job.task), WAIT)
+        await asyncio.wait_for(asyncio.shield(job.task), DISK_BOUND_SECONDS)
         return job
 
     async def groups(self, job_id, status=200, **query):
@@ -127,7 +131,7 @@ class DuplicateFilesTestCase(ApplyTestCase):
         plan = await self.review(job.id, [self.keep_of(group, path)])
         body = await self.apply(plan["token"])
         cleanup = self.api._jobs.get(body["job"]["id"])
-        await asyncio.wait_for(asyncio.shield(cleanup.task), WAIT)
+        await asyncio.wait_for(asyncio.shield(cleanup.task), DISK_BOUND_SECONDS)
         return plan, cleanup
 
     async def until(self, predicate):
@@ -178,7 +182,7 @@ class TestFindJob(DuplicateFilesTestCase):
         gate.set()
         body = await (await legacy).text()
         job = self.api._jobs.get(summary["id"])
-        await asyncio.wait_for(asyncio.shield(job.task), WAIT)
+        await asyncio.wait_for(asyncio.shield(job.task), DISK_BOUND_SECONDS)
         events = _sse_payloads(body)
         done = events[-1]
         self.assertEqual(done["type"], "complete")
@@ -241,7 +245,7 @@ class TestFindJob(DuplicateFilesTestCase):
         self.assertTrue((await cancel.json())["cancellation_requested"])
         gate.set()
         job = self.api._jobs.get(job_id)
-        await asyncio.wait_for(asyncio.shield(job.task), WAIT)
+        await asyncio.wait_for(asyncio.shield(job.task), DISK_BOUND_SECONDS)
         self.assertEqual(job.state, "cancelled")
         last = _sse_payloads(await (await stream).text())[-1]
         self.assertEqual(last, {"type": "error", "message": "Duplicate scan cancelled"})
@@ -341,6 +345,8 @@ class TestReview(DuplicateFilesTestCase):
         self.assertEqual(plan["protected"], {"count": 0, "reasons": []})
 
     async def test_cannot_delete_keeper_or_outside_group(self):
+        # About which files may go, not the backup: skip its fsync-heavy copy.
+        PromptManagerConfig.SAFETY_BACKUP = False
         self.trio()
         self.write("e.png", b"other pair")
         self.write("f.png", b"other pair")
@@ -372,7 +378,7 @@ class TestReview(DuplicateFilesTestCase):
         self.assertEqual([row["label"] for row in plan["details"]], ["b.png"])
         body = await self.apply(plan["token"])
         cleanup = self.api._jobs.get(body["job"]["id"])
-        await asyncio.wait_for(asyncio.shield(cleanup.task), WAIT)
+        await asyncio.wait_for(asyncio.shield(cleanup.task), DISK_BOUND_SECONDS)
         self.assertEqual(cleanup.counters["removed"], 1)
         self.assertTrue((self.output_dir / "a.png").is_file())
         self.assertTrue((self.output_dir / "sub" / "c.png").is_file())
@@ -496,7 +502,7 @@ class TestCleanupJob(DuplicateFilesTestCase):
         plan = await self.review(job.id, keep)
         body = await self.apply(plan["token"])
         cleanup = self.api._jobs.get(body["job"]["id"])
-        await asyncio.wait_for(asyncio.shield(cleanup.task), WAIT)
+        await asyncio.wait_for(asyncio.shield(cleanup.task), DISK_BOUND_SECONDS)
         self.assertTrue(kept.is_file())
         self.assertFalse(gone.exists())
         self.assertFalse(video_thumb.exists())
@@ -508,7 +514,7 @@ class TestCleanupJob(DuplicateFilesTestCase):
         self.write("b.png", b"new bytes")  # changed since the review
         body = await self.apply(plan["token"])
         cleanup = self.api._jobs.get(body["job"]["id"])
-        await asyncio.wait_for(asyncio.shield(cleanup.task), WAIT)
+        await asyncio.wait_for(asyncio.shield(cleanup.task), DISK_BOUND_SECONDS)
         self.assertEqual(cleanup.counters, {"removed": 1, "kept": 2, "errors": 0})
         self.assertEqual((self.output_dir / "b.png").read_bytes(), b"new bytes")
         self.assertFalse((self.output_dir / "sub" / "c.png").exists())
@@ -522,7 +528,7 @@ class TestCleanupJob(DuplicateFilesTestCase):
         self.write("a.png", b"edited keeper")
         body = await self.apply(plan["token"])
         cleanup = self.api._jobs.get(body["job"]["id"])
-        await asyncio.wait_for(asyncio.shield(cleanup.task), WAIT)
+        await asyncio.wait_for(asyncio.shield(cleanup.task), DISK_BOUND_SECONDS)
         self.assertEqual(cleanup.counters["removed"], 0)
         self.assertTrue((self.output_dir / "b.png").is_file())
         self.assertTrue((self.output_dir / "sub" / "c.png").is_file())
@@ -549,7 +555,7 @@ class TestCleanupJob(DuplicateFilesTestCase):
         with patch.object(duplicate_files, "_unlink", side_effect=unlink):
             body = await self.apply(plan["token"])
             cleanup = self.api._jobs.get(body["job"]["id"])
-            await asyncio.wait_for(asyncio.shield(cleanup.task), WAIT)
+            await asyncio.wait_for(asyncio.shield(cleanup.task), DISK_BOUND_SECONDS)
         self.assertEqual(cleanup.state, "failed")
         self.assertEqual(cleanup.counters, {"removed": 1, "kept": 1, "errors": 1})
         self.assertEqual(
@@ -584,7 +590,7 @@ class TestCleanupJob(DuplicateFilesTestCase):
         ):
             body = await self.apply(plan["token"])
             cleanup = self.api._jobs.get(body["job"]["id"])
-            await asyncio.wait_for(asyncio.shield(cleanup.task), WAIT)
+            await asyncio.wait_for(asyncio.shield(cleanup.task), DISK_BOUND_SECONDS)
         self.assertEqual(cleanup.counters, {"removed": 0, "kept": 3, "errors": 1})
         self.assertEqual(
             cleanup.outcomes["errors"][0]["detail"], duplicate_files.DELETE_FAILED
@@ -619,7 +625,7 @@ class TestCleanupJob(DuplicateFilesTestCase):
             with context:
                 body = await self.apply(plan["token"])
                 cleanup = self.api._jobs.get(body["job"]["id"])
-                await asyncio.wait_for(asyncio.shield(cleanup.task), WAIT)
+                await asyncio.wait_for(asyncio.shield(cleanup.task), DISK_BOUND_SECONDS)
         finally:
             if handle is not None:  # pragma: no cover - Windows
                 handle.close()
@@ -642,7 +648,7 @@ class TestCleanupJob(DuplicateFilesTestCase):
         with patch.object(duplicate_files, "_unlink", side_effect=unlink):
             body = await self.apply(plan["token"])
             cleanup = self.api._jobs.get(body["job"]["id"])
-            await asyncio.wait_for(asyncio.shield(cleanup.task), WAIT)
+            await asyncio.wait_for(asyncio.shield(cleanup.task), DISK_BOUND_SECONDS)
         self.assertEqual(cleanup.counters["removed"], 2)
         self.assertEqual(open_at_unlink, [])
 
@@ -703,7 +709,7 @@ class TestCleanupJob(DuplicateFilesTestCase):
         with patch.object(duplicate_files, "_unlink", side_effect=unlink):
             body = await self.apply(plan["token"])
             cleanup = self.api._jobs.get(body["job"]["id"])
-            await asyncio.wait_for(asyncio.shield(cleanup.task), WAIT)
+            await asyncio.wait_for(asyncio.shield(cleanup.task), DISK_BOUND_SECONDS)
         # Its records moved to the identical kept copy before the unlink failed.
         [image] = self.api.db.get_prompt_images(prompt)
         self.assertTrue(image["image_path"].endswith("a.png"))
@@ -719,7 +725,7 @@ class TestCleanupJob(DuplicateFilesTestCase):
         self.assertEqual((busy["code"], busy["running"]["id"]), ("busy", find["id"]))
         gate.set()
         await asyncio.wait_for(
-            asyncio.shield(self.api._jobs.get(find["id"]).task), WAIT
+            asyncio.shield(self.api._jobs.get(find["id"]).task), DISK_BOUND_SECONDS
         )
         # The token survived the refusal; the cleanup holds off a new find.
         release = asyncio.Event()
@@ -737,7 +743,7 @@ class TestCleanupJob(DuplicateFilesTestCase):
             self.assertEqual(legacy.status, 409)
             release.set()
             cleanup = self.api._jobs.get(body["job"]["id"])
-            await asyncio.wait_for(asyncio.shield(cleanup.task), WAIT)
+            await asyncio.wait_for(asyncio.shield(cleanup.task), DISK_BOUND_SECONDS)
 
     async def test_cleanup_refused_up_front_when_the_backup_does_not_fit(self):
         from py.api import maintenance
@@ -965,7 +971,7 @@ class TestReviewHelpers(DuplicateFilesTestCase):
         GalleryConfig.MONITORING_DIRECTORIES = [self._root(self.second_source())]
         body = await self.apply(plan["token"])
         cleanup = self.api._jobs.get(body["job"]["id"])
-        await asyncio.wait_for(asyncio.shield(cleanup.task), WAIT)
+        await asyncio.wait_for(asyncio.shield(cleanup.task), DISK_BOUND_SECONDS)
         self.assertEqual(cleanup.state, "failed")
         self.assertEqual(cleanup.last_event["code"], "sources-changed")
         self.assertEqual(self.backups(), [])
@@ -985,7 +991,7 @@ class TestReviewHelpers(DuplicateFilesTestCase):
         ):
             body = await self.apply(plan["token"])
             cleanup = self.api._jobs.get(body["job"]["id"])
-            await asyncio.wait_for(asyncio.shield(cleanup.task), WAIT)
+            await asyncio.wait_for(asyncio.shield(cleanup.task), DISK_BOUND_SECONDS)
         self.assertEqual(cleanup.state, "failed")
         self.assertEqual(
             cleanup.outcomes["errors"][-1]["detail"], duplicate_files.RECORDS_FAILED

tests/test_api_duplicate_safety.py

@@ -19,17 +19,17 @@ from pathlib import Path
 from unittest.mock import patch
 
 try:
+    from tests.disk_waits import DISK_BOUND_SECONDS
     from tests.open_handles import open_handles_under
     from tests import test_api_duplicate_files as dup_tests
 except ImportError:  # discovered with tests/ as the top-level directory
     import test_api_duplicate_files as dup_tests
+    from disk_waits import DISK_BOUND_SECONDS
     from open_handles import open_handles_under
 
 from py.api import duplicate_files  # noqa: E402
 from py.config import GalleryConfig  # noqa: E402
 
-WAIT = 10
-
 
 def _locator(path):
     return os.path.normcase(os.path.realpath(str(path)))
@@ -49,7 +49,7 @@ class SafetyTestCase(dup_tests.DuplicateFilesTestCase):
     async def apply_plan(self, plan):
         body = await self.apply(plan["token"])
         cleanup = self.api._jobs.get(body["job"]["id"])
-        await asyncio.wait_for(asyncio.shield(cleanup.task), WAIT)
+        await asyncio.wait_for(asyncio.shield(cleanup.task), DISK_BOUND_SECONDS)
         return cleanup
 
 

tests/test_api_history.py

@@ -1169,3 +1169,47 @@ class HistoryHostCleanupTests(unittest.IsolatedAsyncioTestCase):
                     fence.validate(epoch, generation)
                 with self.assertRaises(sqlite3.Error):
                     api._history_fence()
+
+
+class SharedTempFolderRestoreTests(ImageAPITestCase):
+    """#346: a restore survives another suite starting in the shared temp folder.
+
+    Every ``PromptManagerAPI`` sweeps its database's folder for stages and
+    uploads an interrupted restore left. A fixture database directly in the
+    shared temp folder let any other test process (another worktree's suite,
+    a pre-push hook) delete this restore's files mid-request. The shared
+    folder here is a scratch stand-in for the system temp folder, so the test
+    never sweeps the real one.
+    """
+
+    async def get_application(self):
+        shared = tempfile.TemporaryDirectory(ignore_cleanup_errors=True)
+        # Cleanups run last-in first-out: this one after the fixture's own.
+        self.addCleanup(shared.cleanup)
+        self.shared_tmp = shared.name
+        with patch.object(tempfile, "tempdir", shared.name):
+            return await super().get_application()
+
+    async def test_restore_survives_another_suite_sweeping_the_shared_folder(self):
+        from database import restore_stage
+        from database.models import PromptModel
+
+        self.api.db.save_prompt("kept by the upload")
+        source = self.comfy_dir / "upload.db"
+        self.assertTrue(self.api.db.model.backup_database(str(source)))
+        stage = PromptModel.stage_restore
+        swept = []
+
+        def stage_then_other_suite_starts(model, src_path):
+            staged = stage(model, src_path)
+            # What another suite's PromptManagerAPI does at startup.
+            other = os.path.join(self.shared_tmp, "other-suite.db")
+            swept.extend(restore_stage.sweep_restore_temps(other))
+            return staged
+
+        form = FormData()
+        form.add_field("database_file", source.read_bytes(), filename="upload.db")
+        with patch.object(PromptModel, "stage_restore", stage_then_other_suite_starts):
+            response = await self.client.post("/prompt_manager/restore", data=form)
+        self.assertEqual(response.status, 200, await response.text())
+        self.assertEqual(swept, [], "another suite's sweep reached this restore")

tests/test_api_images.py

@@ -42,6 +42,7 @@ from py import lora_utils  # noqa: E402
 from py import lora_manager_layout as layout_module  # noqa: E402
 from py.api.thumbnail_settings import ThumbnailSettings  # noqa: E402
 from py.api import PromptManagerAPI  # noqa: E402
+from tests.private_temp import private_temp_db  # noqa: E402
 from py.api.prompts import MAX_PAGE_LIMIT  # noqa: E402
 from py.config import GalleryConfig  # noqa: E402
 from utils.hashing import generate_prompt_hash  # noqa: E402
@@ -60,8 +61,7 @@ class ImageAPITestCase(AioHTTPTestCase):
     """App with PromptManager routes, a temp DB and a temp output directory."""
 
     async def get_application(self):
-        self._temp_db = tempfile.NamedTemporaryFile(delete=False, suffix=".db")
-        self._temp_db.close()
+        self._temp_db = private_temp_db(self)
 
         self._output_tmp = tempfile.TemporaryDirectory(ignore_cleanup_errors=True)
         # A fake ComfyUI tree: <tmp>/output is the gallery root and <tmp> is the

tests/test_api_lora.py

@@ -36,6 +36,7 @@ from aiohttp.test_utils import AioHTTPTestCase
 
 from database.operations import PromptDatabase
 from py.api import PromptManagerAPI
+from tests.private_temp import private_temp_db
 from py import config_store, lora_import, lora_utils
 from py.config import IntegrationConfig, PromptManagerConfig
 from py.api import settings_routes as settings_module
@@ -57,8 +58,7 @@ class LoraAPITestCase(AioHTTPTestCase):
     """Stands up the PromptManager routes with a temp DB and neutral LoRA config."""
 
     async def get_application(self):
-        self._temp_db = tempfile.NamedTemporaryFile(delete=False, suffix=".db")
-        self._temp_db.close()
+        self._temp_db = private_temp_db(self)
         self.tmp_root = Path(tempfile.mkdtemp())
         self.custom_nodes = self.tmp_root / "custom_nodes"
         self.custom_nodes.mkdir()

tests/test_api_prompts.py

@@ -5,7 +5,6 @@ Route tests for the prompt API (py/api/prompts.py).
 import json
 import os
 import sys
-import tempfile
 import unittest
 from unittest.mock import MagicMock, patch
 
@@ -29,6 +28,7 @@ except ImportError:  # discovered with tests/ as the top-level directory
 
 from database.operations import PromptDatabase  # noqa: E402
 from py.api import PromptManagerAPI  # noqa: E402
+from tests.private_temp import private_temp_db  # noqa: E402
 import py.api.prompts as prompts_module  # noqa: E402
 from py.api.prompts import (  # noqa: E402
     BULK_CATEGORY_MAX_PROMPTS,
@@ -43,8 +43,7 @@ class PromptAPITestCase(AioHTTPTestCase):
     """App with PromptManager routes and a temporary SQLite database."""
 
     async def get_application(self):
-        self._temp_db = tempfile.NamedTemporaryFile(delete=False, suffix=".db")
-        self._temp_db.close()
+        self._temp_db = private_temp_db(self)
 
         app = web.Application()
         routes = web.RouteTableDef()

tests/test_execution_history_writer.py

@@ -31,6 +31,12 @@ from utils.execution_history_writer import HistoryWriter
 # 0.19 s on macOS py3.13 (PR #351).
 NON_BLOCKING_CEILING = 1.0
 
+# Saturation tests block the worker while they queue 1023 more items. The
+# block must hold until the test releases it (#346/#352): a 3 s timer let a
+# loaded runner unblock the worker mid-fill and drain it. This is only a hang
+# guard; the test, or cleanup, always sets the release.
+SATURATION_HOLD_SECONDS = 60
+
 
 class ManualClock:
     def __init__(self):
@@ -84,6 +90,26 @@ class HistoryWriterTests(unittest.TestCase):
             threading.Event().wait(0.005)
         self.fail("finite deadline expired waiting for worker behavior")
 
+    def wait_for_worker_pass(self, writer):
+        """Wait until the idle worker has run one whole loop pass from now.
+
+        A worker with no accepted work ends every pass by parking on its wake
+        event. The first park counted may end a pass already under way; the
+        second ends a pass that began after this call, so whatever the worker
+        decides at the current clock value has happened (state, not a
+        wall-clock pause that a loaded runner can outlast).
+        """
+        wake = writer._wake
+        park = wake.wait
+        parks = []
+
+        def counted(timeout=None):
+            parks.append(timeout)
+            return park(timeout)
+
+        with patch.object(wake, "wait", counted):
+            self.wait_for(lambda: len(parks) >= 2)
+
     def writer(self, factory=None, ready=True):
         writer = HistoryWriter(factory or (lambda: self.db), clock=self.clock)
         self.writers.append(writer)
@@ -301,7 +327,7 @@ class HistoryWriterTests(unittest.TestCase):
 
         def write(command):
             entered.set()
-            release.wait(3)
+            release.wait(SATURATION_HOLD_SECONDS)
             return original(command)
 
         with patch.object(self.db, "write_execution_start", write):
@@ -348,9 +374,12 @@ class HistoryWriterTests(unittest.TestCase):
         first = self.sealed(writer)
         charge = first.lease.byte_count
 
+        left = threading.Event()
+
         def write(*args, **kwargs):
             entered.set()
             release.wait(3)
+            left.set()
             return original(*args, **kwargs)
 
         with patch.object(self.db, "write_execution_contexts_for_finish", write):
@@ -396,8 +425,21 @@ class HistoryWriterTests(unittest.TestCase):
                 before = monotonic()
                 self.assertFalse(writer.close(0.02))
                 elapsed = monotonic() - before
+            # close() waits until its own monotonic deadline, so the lower
+            # bound holds on any runner.
             self.assertGreaterEqual(elapsed, 0.02)
-            self.assertLess(elapsed, 0.2)
+            if early_join:
+                # Manual clock: only the patched join/sleep advance it, so this
+                # exact bound proves early joins do not spin forever.
+                self.assertLess(elapsed, 0.2)
+            else:
+                # Real clock: a loaded runner may run late, but close() must
+                # not wait anywhere near the 3 s the blocked write holds.
+                self.assertLess(elapsed, 0.02 + NON_BLOCKING_CEILING)
+            # Bounded on the real clock too: close() gave up while the
+            # in-flight write was still blocked, instead of waiting it out.
+            self.assertFalse(left.is_set())
+            self.assertTrue(writer.status().alive)
             self.assertEqual(pool.charged_bytes, charge)
             replacement = HistoryWriter(lambda: self.db, context_budget=pool)
             self.writers.append(replacement)
@@ -771,7 +813,7 @@ class HistoryWriterTests(unittest.TestCase):
             self.wait_for(lambda: not writer.status().accepting)
             self.assertFalse(writer.submit(fresh))
             self.clock.advance(4.99)
-            threading.Event().wait(0.08)
+            self.wait_for_worker_pass(writer)
             self.assertEqual(probes, [0])
             self.clock.advance(0.01)
             self.wait_for(lambda: writer.status().accepting)
@@ -794,12 +836,12 @@ class HistoryWriterTests(unittest.TestCase):
         writer = self.writer(factory, ready=False)
         self.wait_for(lambda: len(calls) == 1)
         self.clock.advance(4.99)
-        threading.Event().wait(0.08)
+        self.wait_for_worker_pass(writer)
         self.assertEqual(calls, [0])
         self.clock.advance(0.01)
         self.wait_for(lambda: len(calls) == 2)
         self.clock.advance(4.99)
-        threading.Event().wait(0.08)
+        self.wait_for_worker_pass(writer)
         self.assertEqual(calls, [0, 5])
         self.clock.advance(0.01)
         self.wait_for(lambda: writer.status().accepting)
@@ -853,7 +895,7 @@ class HistoryWriterTests(unittest.TestCase):
                 )
             writer.invalidate()
             release.set()
-            threading.Event().wait(0.08)
+            self.wait_for_worker_pass(writer)
             self.assertFalse(writer.status().accepting)
             self.assertIsNone(writer.status().epoch)
             self.assertFalse(writer.submit(self.start_command(epoch=stale_epoch)))
@@ -871,7 +913,7 @@ class HistoryWriterTests(unittest.TestCase):
         lock.execute("BEGIN IMMEDIATE")
         before = time.monotonic()
         self.assertTrue(writer.submit(command))
-        self.assertLess(time.monotonic() - before, 0.1)
+        self.assertLess(time.monotonic() - before, NON_BLOCKING_CEILING)
         self.wait_for(lambda: not writer.status().accepting)
         lock.rollback()
         self.clock.advance(5)

tests/test_execution_output_writer.py

@@ -18,6 +18,7 @@ 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
+from tests.disk_waits import DISK_BOUND_SECONDS
 
 try:
     from tests import test_execution_history_writer as fixtures
@@ -28,11 +29,17 @@ except ImportError:
 # Upper bound a blocked fake write waits for its release. Tests prove ordering
 # with events, so this only caps a failing run and never decides a pass.
 BLOCK_SECONDS = 10
+
+# Saturation tests block the worker while they queue 1023 more items. The
+# block must hold until the test releases it (#346/#352): a 3 s timer let a
+# loaded runner unblock the worker mid-fill and drain it. This is only a hang
+# guard; the test, or cleanup, always sets the release.
+SATURATION_HOLD_SECONDS = 60
+
 # A restore is a full verify/migrate/backup/replace cycle and is bound by the
 # disk: a busy Windows runner took over 10 s (merge queue, #382). Waits that
-# span a restore use this ceiling, which stays under the runner's 120 s hang
-# watchdog, so a real hang still fails with the thread dump.
-RESTORE_SECONDS = 90
+# span a restore use the shared disk-bound hang guard (tests/disk_waits.py).
+RESTORE_SECONDS = DISK_BOUND_SECONDS
 
 
 class OutputWriterTests(fixtures.unittest.TestCase):
@@ -591,7 +598,7 @@ class OutputWriterTests(fixtures.unittest.TestCase):
 
         def write(command):
             entered.set()
-            release.wait(3)
+            release.wait(SATURATION_HOLD_SECONDS)
             return original(command)
 
         with patch.object(self.db, "write_execution_start", write):

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_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_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"},

tests/test_workspace_system_fixture.py

@@ -1,12 +1,26 @@
-"""The System browser hooks only ever reset their own server's config.json (#249 S3)."""
+"""The System browser hooks only ever reset their own server's config.json (#249 S3)
+and decide JPEG XL availability whatever the host has installed (#371)."""
 
+import asyncio
 import os
+import sys
 import tempfile
+import types
 import unittest
 from pathlib import Path
 from unittest import mock
 
-from tests.browser.workspace_system_fixture import register_system_commands
+from aiohttp import web
+from aiohttp.test_utils import TestClient, TestServer
+
+_mock_server = mock.MagicMock()
+_mock_server.PromptServer.instance.routes = mock.MagicMock()
+sys.modules.setdefault("server", _mock_server)
+
+from py.api import thumbnail_encode  # noqa: E402
+from tests.browser.workspace_system_fixture import (  # noqa: E402
+    register_system_commands,
+)
 
 
 class TestWorkspaceSystemFixture(unittest.TestCase):
@@ -33,6 +47,48 @@ class TestWorkspaceSystemFixture(unittest.TestCase):
                     register_system_commands(None, None, None, self.root)
         self.assertEqual(elsewhere.read_bytes(), b'{"real": true}')
 
+    def test_no_encoder_holds_even_when_the_host_has_one(self):
+        """``{available: false}`` survives the settings read's re-probe (#371)."""
+        config = self.root / "config.json"
+        env = {**os.environ, "PROMPT_MANAGER_CONFIG_PATH": str(config)}
+        api = types.SimpleNamespace()
+        app, routes = web.Application(), web.RouteTableDef()
+        real_detect = thumbnail_encode._detect
+        with mock.patch.dict(os.environ, env, clear=True):
+            reset = register_system_commands(app, routes, api, self.root)
+        app.add_routes(routes)
+        self.addCleanup(thumbnail_encode.reset_encoders)
+        self.addCleanup(setattr, thumbnail_encode, "_detect", real_detect)
+
+        async def scenario():
+            async with TestClient(TestServer(app)) as client:
+
+                async def post(available):
+                    response = await client.post(
+                        "/__test__/system/jxl-encoder", json={"available": available}
+                    )
+                    self.assertEqual(response.status, 200)
+
+                await post(False)
+                status = thumbnail_encode.encoder_status()["jxl"]
+                self.assertFalse(status["available"])
+                self.assertIsNone(
+                    thumbnail_encode.detect_jxl_encoder(recheck_missing=True)
+                )
+                await post(True)
+                status = thumbnail_encode.encoder_status()["jxl"]
+                self.assertEqual(status["encoder"], "cjxl")
+                await post(False)
+
+        # The host's real detector finds an encoder (Pillow can save JPEG XL).
+        with mock.patch.object(thumbnail_encode, "_pillow_can_save_jxl", lambda: True):
+            asyncio.run(scenario())
+            # Reset gives detection back to the host.
+            reset()
+            self.assertIs(thumbnail_encode._detect, real_detect)
+            status = thumbnail_encode.encoder_status()["jxl"]
+            self.assertEqual(status["encoder"], "pillow")
+
 
 if __name__ == "__main__":
     unittest.main()

tests/vendor/ci-workflow.test.mjs

@@ -247,3 +247,32 @@ test('the Python matrix and coverage run the suite through the watchdog runner',
         assert.doesNotMatch(found.get(id), /unittest discover/, `${id} no longer runs the bare unittest discover`);
     }
 });
+
+// CI hygiene (#352): lint tools are pinned, the bandit hook can fail, the JS job caches npm.
+const readRepoFile = path => readFileSync(new URL(`../../${path}`, import.meta.url), 'utf8');
+
+test('black and flake8 are pinned exactly and the lint job installs those pins', () => {
+    const requirements = readRepoFile('requirements-dev.txt');
+    assert.match(requirements, /^black==\d+\.\d+\.\d+$/m, 'black is pinned exactly');
+    assert.match(requirements, /^flake8==\d+\.\d+\.\d+$/m, 'flake8 is pinned exactly');
+    const lint = jobs(readWorkflow('test.yml')).get('lint');
+    assert.match(lint, /pip install \$\(grep -E '\^\(black\|flake8\)==' requirements-dev\.txt\)/);
+    assert.doesNotMatch(lint, /pip install black flake8/, 'lint no longer installs the latest releases');
+});
+
+test('the frontend modules job caches npm keyed on the vendor lockfile', () => {
+    const setupNode = steps(jobs(readWorkflow('test.yml')).get('js-test'))
+        .find(step => /uses: actions\/setup-node@v\d+/.test(step));
+    assert.match(setupNode, /^ {10}cache: npm$/m);
+    assert.match(setupNode, /^ {10}cache-dependency-path: tools\/vendor\/package-lock\.json$/m);
+});
+
+test('the bandit pre-commit hook fails at the severity CI enforces and passes every file', () => {
+    const quality = readWorkflow('code-quality.yml');
+    assert.match(quality, /^ {8}run: bandit -r \. -lll -q --exclude /m, 'CI fails on high severity');
+    const hook = /- id: bandit\n([\s\S]*?)(?=\n {6}(?:- id:|#)|\n*$)/.exec(readRepoFile('.pre-commit-config.yaml'))?.[1];
+    assert.ok(hook, 'the bandit hook exists');
+    assert.match(hook, /^ {8}entry: bandit -lll -q$/m, 'the hook enforces the same -lll level as CI');
+    assert.doesNotMatch(hook, /true|exit-zero|bash -c/, 'nothing masks the exit code or drops a file');
+    assert.doesNotMatch(hook, /pass_filenames: false/, 'the hook scans the staged files');
+});

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

web/js/workspace/images-state.mjs

@@ -275,8 +275,13 @@ export function createImageStream({ api, onChange = () => {}, batchSize = 40, re
     };
 }
 
-/** The concept's count line (app.js a446): "N of M images · K positive prompts", without the count while unknown. */
-export function countLine({ loaded, total, positiveCount }) {
+/**
+ * The concept's count line (app.js a446): "N of M images · K positive prompts", without the count while unknown, and
+ * the page's first "Loading images…" while a first page loads.
+ */
+export function countLine({ loaded, total, positiveCount, loading = false, error = null }) {
+    // A first page in flight has no count yet: "0 of 0 images" would claim the filter matched nothing (#391).
+    if (loading && !loaded && !error) return 'Loading images…';
     const images = `${loaded.toLocaleString('en-US')} of ${total.toLocaleString('en-US')} ${noun(total, 'image')}`;
     return isCount(positiveCount) ? `${images} · ${positiveCount.toLocaleString('en-US')} ${noun(positiveCount, 'positive prompt')}` : images;
 }

web/js/workspace/workspace.css

@@ -433,10 +433,13 @@ pre.prompt-text { font: inherit; font-size: 12px; line-height: 1.75; margin-bloc
 }
 /* D132: at phone width the concept's 120px cap cuts every engine label ("WD14 SwinV2 · ~400 MB"); the Tagging model
    select alone keeps its natural width there (its widest engine label) and, with no room beside the row text, wraps
-   under it. */
+   under it. Existing tags (#398) never shrinks for its row text and, past the cap, keeps its natural width beside the
+   text ("Replace after review"), bounded by its row like the engine select. */
+.setting-row #tag-existing { flex-shrink: 0; /* D132 */ }
 @media (max-width: 700px) {
   #workspace-main .setting-row:has(> #tag-model) { flex-wrap: wrap; /* D132 */ }
   #workspace-main .setting-row #tag-model { flex-shrink: 0; max-width: 100% !important; /* D132 */ }
+  #workspace-main .setting-row #tag-existing { max-width: 100% !important; /* D132 */ }
 }
 
 /* D109: System › Diagnostics & logs' added Lines field sits in the concept's toolbar like its Level select, and a value