Merge remote-tracking branch 'origin/main' into fix/376-exit-cancel

# Conflicts:
#	utils/parallel.py

TESTING.md

@@ -313,6 +313,26 @@ from the tests' package unload (`cleanup_package`), and it removes its own
 ComfyUI-Manager restart through `os.execv`, a closed Windows console, or
 ComfyUI Desktop stopping its server. SQLite's WAL journal covers those.
 
+Jobs running at exit (#376): before any `atexit` callback, Python joins every
+executor thread (`concurrent.futures` registers that join with
+`threading._register_atexit`). The package registers its own callback there
+(`utils/exit_signal.install`, after `concurrent.futures`, so it runs first):
+it sets the exit signal and kills the running encoder processes
+(`shutdown_encoders`, the ffmpeg that never sees Ctrl+C on Windows). The call in
+flight then stops at its next safe point: `Job.cancelled` is True, so every
+check a job already makes between files stops it, `map_parallel` starts no
+further item (`ExitRequested`), the scan walk stops between folders, LoRA
+import staging stops before its next sidecar (also for the legacy scan) and a
+tagging model download stops before its next chunk where huggingface_hub takes
+`tqdm_class` (1.x removes its temp file; 0.x `hf_hub_download` does not take it
+and finishes the file). Xet transfers ignore that checkpoint, so the hook also
+calls hub's own `abort_xet_session` when huggingface_hub is loaded. The
+trigger-word cache load (small JSON sidecars) has no checkpoint. No check sits
+inside a library write. That hook is
+CPython-internal; without it (or when it is refused) `install` returns False and
+exit waits for the call in flight, as before. `tests/test_exit_cancel.py` runs a
+child process with three such calls running and requires its exit within 15 s.
+
 Stop contract: `PromptTracker.shutdown(timeout)`,
 `ImageGenerationHandler.shutdown(timeout)` (restartable),
 `ImageGenerationHandler.close(timeout)` (final: `schedule()` refuses and a
@@ -3983,7 +4003,30 @@ modals use the same routes and keep working.
   other: each file's identity is checked just before its unlink.
 - **Clear all logs** (`POST /logs/truncate`, also the admin page's Clear) empties the
   active log through its handler, deletes the rotated copies (Rule R) and clears the
-  buffer; other files and links are kept. Response keys are unchanged.
+  buffer; other files and links are kept. Response keys are unchanged. It deletes or
+  empties no more than its confirmation showed (#389):
+  - `/logs/files` rows carry `identity`, the string `"<size>:<mtime_ns>:<ino>"` (a
+    string: `mtime_ns` and many inodes are beyond a browser's exact integers). The
+    workspace sends `{"expected": [{name, identity}]}` with the active and rotated rows
+    its dialog was built from. Under the files lock and every handler lock each listed
+    rotated copy must still match (`_matches_review`) and the listed active log must be
+    the same inode, or nothing changes: 409 `logs-changed`. Only listed copies are
+    deleted; one that appeared after the confirmation is kept. A malformed list is 400.
+  - Without a body (the legacy admin page, whose confirm lists nothing; also the
+    workspace when the file list could not be read and the dialog said "deletes its
+    rotated copies") every rotated copy present at the locked rescan is deleted.
+  - Either way, before anything changes (best effort, not a cross-process lock: a
+    write before our own last write, or any while file logging is off here, is not
+    seen): a second writer of the active log (our
+    handler's flushed stream is behind the file's size, or its file is now a rotated
+    copy: another ComfyUI sharing the folder) is 409 `logs-foreign-writer`; an active
+    log with more than one name (`st_nlink > 1` by `lstat` or the handler's `fstat`,
+    NTFS included) is 409 `logs-hard-link`, never truncated (replacing it instead would
+    rename a file our handler holds open, which Windows refuses). Each copy is
+    `lstat`-checked again just before its unlink; one that changed stops the clear there
+    and keeps it, the copies after it and the active log, which is emptied last.
+  - The browser fixture grows the active log through the handler's own stream, since a
+    second handle appending to it is exactly what Clear refuses.
 - **Runtime settings** (`GET/POST /logs/config`): the level (DEBUG…CRITICAL) and console
   logging apply until the server restarts; the retention values are shown read-only in
   the workspace and the API keeps accepting them.
@@ -4623,7 +4666,23 @@ files, oldest forgotten first, keyed by 16-byte digests: 8.6 MiB measured at
 the cap). A file whose inspection raises is logged, counted in
 `review_errors` and reported `unresolved`; the scan goes on. An older version's
 pending, undecided item of bytes now untraced leaves the queue
-(`settle_stale_review_items`), so an item never stays stuck at 409. The 16-candidate limited
+(`settle_stale_review_items`), so an item never stays stuck at 409.
+Saver anchor (#390, `tests/test_review_saver_anchor.py`): the scan and Apply
+resolve with the file's name as the workspace inspector does, so the one saver
+whose `filename_prefix` names the file anchors its branch. A file is never both
+unlinked and unqueued (maintainer ruling): when only that anchor resolves it
+(the named inspection run resolves, the unnamed one does not) and neither the
+default profile nor the parser floor links it, the scan queues it with the
+anchor's pair as its one candidate (evidence `anchor: "saver"`; the page
+preselects it), never linking it. Such an item counts as a choice for the
+untraced settle. Files already linked or resolved by the default path are
+unchanged. No match, several matches or an unfinished walk anchor nothing, and
+the file is queued as before. `REVIEW_RULE_VERSION` is folded into inspection
+item versions, so older inspection items are recorded again once. Apply accepts
+the anchored pair and refuses a choice the anchor contradicts. Measured on the
+maintainer's 40,890 outputs: the queue is unchanged (249 files); the 242
+`gpu_unlock_bench` files the inspector resolves are linked by the parser
+floor. The 16-candidate limited
 rule and the 8 MiB bound apply unchanged. The queue heals itself: after each batch, the scan removes the pending
 items of every listed file (read or skipped as known) that a saved row now
 links, and the monitor does the same for each ambiguous output it links

__init__.py

@@ -355,6 +355,44 @@ async def _stop_background_work_on_cleanup(app):
     _stop_background_work()
 
 
+def _kill_encoders_at_exit():
+    """Run by the exit hook before Python joins the executor threads (#376):
+    a JPEG XL or ffmpeg encoder (no console on Windows, so Ctrl+C never
+    reaches it) is killed instead of being waited for."""
+    from .py.api.thumbnail_encode import shutdown_encoders
+
+    shutdown_encoders()
+
+
+def _abort_downloads_at_exit():
+    """Run by the exit hook (#376): huggingface_hub's Xet transfers ignore
+    the chunk checkpoint, so they are aborted the way hub aborts on Ctrl+C."""
+    import sys
+
+    if "huggingface_hub" not in sys.modules:
+        return  # no download ever ran: import nothing at exit
+    from .py.autotag import abort_hub_downloads
+
+    abort_hub_downloads()
+
+
+try:
+    # Before the executor join: running jobs stop at their next file, item or
+    # chunk (#376). The teardown below still runs from atexit afterwards.
+    from .utils import exit_signal as _exit_signal
+
+    _exit_signal.install(_kill_encoders_at_exit)
+    _exit_signal.install(_abort_downloads_at_exit)
+except Exception as e:
+    try:
+        from .utils.logging_config import get_logger
+
+        get_logger("prompt_manager.init").error(
+            f"Failed to register the exit signal: {e}"
+        )
+    except Exception:
+        pass
+
 try:
     atexit.register(_stop_background_work)
     from .py.config import server_instance as _teardown_server

database/review_schema.py

@@ -229,13 +229,15 @@ def evidence_offers_choice(evidence: Any) -> bool:
     """
     if not isinstance(evidence, dict):
         return False
+    # The saver anchor's one pair (#390) is a suggestion to confirm.
+    needed = 1 if evidence.get("anchor") == "saver" else MIN_CHOOSABLE
     if evidence.get("limited"):
         total = evidence.get("candidate_total")
-        return type(total) is int and total >= MIN_CHOOSABLE
+        return type(total) is int and total >= needed
     candidates = evidence.get("candidates")
     if not isinstance(candidates, list):
         return False
-    return sum(1 for c in candidates if choosable(c)) >= MIN_CHOOSABLE
+    return sum(1 for c in candidates if choosable(c)) >= needed
 
 
 def evidence_digest(text: str) -> str:

py/api/admin.py

@@ -34,6 +34,11 @@ try:
 except ImportError:
     from utils.parallel import map_parallel
 
+try:
+    from ...utils import exit_signal
+except ImportError:
+    from utils import exit_signal
+
 try:
     from ...database.pair_ingestion import blank_positive, ingest_resolved_pair
     from ...database.run_attribution import (
@@ -118,6 +123,7 @@ def _collect_media_files(output_dirs, extensions):
         if not os.path.isdir(root):
             continue
         for current, dirnames, filenames in os.walk(root, followlinks=False):
+            exit_signal.checkpoint()  # #376: stop between directories at exit
             depth = 0 if current == root else len(Path(current).relative_to(root).parts)
             dirnames[:] = sorted(
                 d
@@ -1630,7 +1636,14 @@ class AdminRoutesMixin:
                 if self._review_file_linked(media_file, roots):
                     outcome = "unchanged"  # already linked: not a review case
                 else:
-                    self._record_scanned_review(media_file, roots, review)
+                    # The records name the item versions that count; the
+                    # saver anchor (#390) is checked on the captured bytes.
+                    self._record_scanned_review(
+                        media_file,
+                        roots,
+                        review,
+                        scanned=(records.get("canonical"), None),
+                    )
             elif untraced and outcome == "unresolved":
                 # Untraced by the default profile: the inspection profile may
                 # still offer a choice (queued for review, never linked).
@@ -1691,9 +1704,9 @@ class AdminRoutesMixin:
         ``"ambiguous"`` (nothing linked it) is a Needs review case. An
         ambiguous result with nothing to choose (``reviewable`` is False: fewer
         than two candidates carry a prompt) that nothing links returns
-        ``"unresolved"`` instead, and ``media_file`` goes to ``untraced``;
-        its parsed records go to ``records["canonical"]`` when ``records`` is
-        a dict.
+        ``"unresolved"`` instead, and ``media_file`` goes to ``untraced``.
+        The parsed records of either ambiguous file go to
+        ``records["canonical"]`` when ``records`` is a dict.
         """
         if not metadata:
             # No prompt metadata (--disable-metadata, bare savers): only the
@@ -1723,6 +1736,8 @@ class AdminRoutesMixin:
                 records["canonical"] = canonical
         elif resolution["state"] == "ambiguous" and ambiguous is not None:
             ambiguous.append(media_file)
+            if records is not None:
+                records["canonical"] = canonical
         if resolution["state"] == "resolved":
             result = ingest_resolved_pair(
                 self.db, str(media_file), canonical, resolution

py/api/jobs.py

@@ -47,6 +47,11 @@ import time
 import uuid
 from pathlib import PurePosixPath, PureWindowsPath
 
+try:
+    from ...utils import exit_signal
+except ImportError:  # tests import the package top-level
+    from utils import exit_signal
+
 logger = logging.getLogger(__name__)
 # Start and finish order: wall-clock times can tie (coarse Windows clocks).
 _sequence = itertools.count()
@@ -195,11 +200,21 @@ class Job:
         self.counters = dict.fromkeys(groups, 0)
         self.outcomes = {group: [] for group in groups}
         self.log = []
-        self.cancelled = False
+        self._cancel_requested = False
         self.last_event = None
         self.task = None
         self._subscribers = set()
 
+    @property
+    def cancelled(self):
+        """Cancel was requested, or the interpreter is exiting (#376): every
+        check a job already makes between files also stops it at exit."""
+        return self._cancel_requested or exit_signal.requested()
+
+    @cancelled.setter
+    def cancelled(self, value):
+        self._cancel_requested = bool(value)
+
     @property
     def record_changing(self):
         return self.admission == "records"

py/api/logging_routes.py

@@ -11,6 +11,11 @@ import stat
 
 from aiohttp import web
 
+try:
+    from ...utils.logging_config import ClearRefused, parse_confirmed
+except ImportError:  # loaded as the top-level ``py`` package (tests)
+    from utils.logging_config import ClearRefused, parse_confirmed
+
 # Keys PromptManagerLogger.update_config() understands. Anything else is
 # rejected so a client cannot inject arbitrary entries into the logging config.
 LOG_CONFIG_KEYS = frozenset(
@@ -263,10 +268,43 @@ class LoggingRoutesMixin:
             )
 
     async def truncate_logs(self, request):
-        """Truncate all log files."""
+        """Clear all logs: what ``{"expected": [...]}`` lists, or all (#389).
+
+        ``expected`` holds the ``{name, identity}`` rows of ``/logs/files`` the
+        confirmation showed. Without a body (the legacy admin page, whose
+        confirmation lists nothing) every rotated copy present now is deleted.
+        A malformed list is 400; a clear that would empty or delete what was
+        not confirmed changes nothing and is 409 with the reason's ``code``.
+        """
         try:
+            args = ()
+            if request.can_read_body:
+                data, error_response = await self._read_json_body(request)
+                if error_response is not None:
+                    return error_response
+                if "expected" in data:
+                    try:
+                        parse_confirmed(data["expected"])
+                    except ValueError:
+                        return web.json_response(
+                            {
+                                "success": False,
+                                "error": "expected must list name and identity"
+                                " rows from /logs/files",
+                            },
+                            status=400,
+                        )
+                    args = (data["expected"],)
             logger_manager = self._get_logger_manager()
-            results = await self._run_in_executor(logger_manager.truncate_logs)
+            try:
+                results = await self._run_in_executor(
+                    logger_manager.truncate_logs, *args
+                )
+            except ClearRefused as refused:
+                return web.json_response(
+                    {"success": False, "error": str(refused), "code": refused.code},
+                    status=409,
+                )
 
             return web.json_response(
                 {

py/api/review.py

@@ -677,7 +677,14 @@ def _verify_and_read(store, capture, item, root, need_metadata: bool):
     facts = store.verify(root, relative)
     captured = None
     if need_metadata:
-        captured = captured_review(store, capture, item_profile(item["evidence"]))
+        # The file's own name, so an inspection item resolves anchored on its
+        # saver as it was recorded (#390).
+        captured = captured_review(
+            store,
+            capture,
+            item_profile(item["evidence"]),
+            relative.rsplit("/", 1)[-1],
+        )
     return relative, facts, captured
 
 
@@ -723,6 +730,9 @@ def _candidate_still_offered(resolution: dict, choice: str) -> bool:
     that rewords the top-level reason or reorders candidates leaves a
     pending item applicable.
     """
+    if isinstance(resolution, dict) and resolution.get("anchor") == "saver":
+        # Only the saver anchor resolves the bytes (#390): its pair alone.
+        resolution = {"state": "ambiguous", "candidates": [resolution]}
     if not isinstance(resolution, dict) or resolution.get("state") != "ambiguous":
         return False
     evidence, _count, limited = review_evidence(resolution)

py/api/review_recording.py

@@ -88,7 +88,8 @@ class ReviewRecordingMixin:
     def _record_scanned_review(self, media_file, roots, review=None, scanned=None):
         """Record one ambiguous, unlinked scanned file (logged and counted).
 
-        ``scanned`` is the scan's ``(metadata, inspection_resolution)``.
+        ``scanned`` is the scan's ``(metadata, inspection_resolution)``; the
+        resolution is None when the scan did not run the inspection profile.
         """
         try:
             outcome = self._record_review_sync(
@@ -199,7 +200,7 @@ class ReviewRecordingMixin:
         key = _memory_key(locator)
         if state is not None and memory.get(key) == state:
             return None
-        resolution = review_capture.inspection_choice(canonical)
+        resolution = review_capture.inspection_choice(canonical, media_file.name)
         if resolution is not None:
             memory.pop(key, None)
             return canonical, resolution

py/api/thumbnail_auto.py

@@ -92,6 +92,11 @@ try:
 except ImportError:  # tests import the package top-level
     from utils.parallel import map_parallel, note_refusal, refused_at_shutdown
 
+try:
+    from ...utils.exit_signal import ExitRequested
+except ImportError:  # tests import the package top-level
+    from utils.exit_signal import ExitRequested
+
 # The placeholder frame a video without a preview is answered with (JPEG
 # bytes, never written to disk); written previews follow the settings.
 PLACEHOLDER_SIZE = (thumbnail_settings.DEFAULT_EDGE, thumbnail_settings.DEFAULT_EDGE)
@@ -460,6 +465,8 @@ class AutoThumbnailer:
     def _run(self):
         try:
             self._drain()
+        except ExitRequested:
+            pass  # #376: the interpreter is exiting; queued files are dropped
         except Exception:
             self._logger.warning("The automatic preview worker stopped", exc_info=True)
         finally:

py/autotag.py

@@ -7,6 +7,7 @@ Refactored from standalone_tagger.py for integration with PromptManager API.
 
 import errno
 import gc
+import inspect
 import os
 import sys
 from pathlib import Path
@@ -32,6 +33,54 @@ try:
 except ImportError:
     from py.tagging_defaults import DEFAULT_TAG_TEMPLATE
 
+try:
+    from ..utils import exit_signal
+except ImportError:
+    from utils import exit_signal
+
+
+def abort_hub_downloads() -> None:
+    """Abort huggingface_hub's Xet transfers (run by the exit hook, #376).
+
+    Xet downloads report progress from native code, which ignores an
+    exception raised in ``tqdm.update``; huggingface_hub aborts them itself
+    on Ctrl+C with ``abort_xet_session``. Nothing happens when
+    huggingface_hub was never imported or has no Xet support.
+    """
+    if "huggingface_hub" not in sys.modules:
+        return
+    try:
+        from huggingface_hub.utils import _xet
+    except ImportError:
+        return
+    abort = getattr(_xet, "abort_xet_session", None)
+    if callable(abort):
+        abort()
+
+
+def _stop_at_exit(download) -> Dict[str, Any]:
+    """Keyword arguments that stop a huggingface_hub download at its next
+    chunk once the interpreter is exiting (#376), or none when this
+    huggingface_hub does not take ``tqdm_class``.
+
+    huggingface_hub 1.x removes its partial temp file on any failure; older
+    versions keep their resumable ``.incomplete`` file. The model's own path
+    is only written by the final rename either way.
+    """
+    try:
+        if "tqdm_class" not in inspect.signature(download).parameters:
+            return {}
+        from huggingface_hub.utils import tqdm as hub_tqdm
+    except (ImportError, TypeError, ValueError):
+        return {}
+
+    class StopAtExit(hub_tqdm):
+        def update(self, n=1):
+            exit_signal.checkpoint()
+            return super().update(n)
+
+    return {"tqdm_class": StopAtExit}
+
 
 def _autotag_config():
     """The saved defaults; imported late, as config needs ComfyUI's server."""
@@ -397,6 +446,7 @@ class AutoTagService:
                     filename=filename,
                     local_dir=str(model_dir),
                     local_dir_use_symlinks=False,
+                    **_stop_at_exit(hf_hub_download),
                 )
 
         if progress_callback:
@@ -541,6 +591,13 @@ class AutoTagService:
             else:
                 return self._download_hf_model(progress_callback)
         except Exception as e:
+            if exit_signal.requested():
+                # ExitRequested between chunks, or the Xet session aborted
+                # by the exit hook (abort_hub_downloads).
+                self.logger.info(
+                    f"Model download stopped: ComfyUI is exiting ({model_type})"
+                )
+                return False
             self.logger.error(f"Download failed: {e}")
             if progress_callback:
                 progress_callback(f"Error: {str(e)}", 0)
@@ -570,6 +627,7 @@ class AutoTagService:
                 filename=config["filename"],
                 local_dir=str(gguf_dir),
                 local_dir_use_symlinks=False,
+                **_stop_at_exit(hf_hub_download),
             )
             self.logger.info("GGUF model downloaded")
 
@@ -587,6 +645,7 @@ class AutoTagService:
                 filename=config["mmproj_filename"],
                 local_dir=str(gguf_dir),
                 local_dir_use_symlinks=False,
+                **_stop_at_exit(hf_hub_download),
             )
             self.logger.info("mmproj downloaded")
 
@@ -613,6 +672,7 @@ class AutoTagService:
                 repo_id=config["repo"],
                 local_dir=str(model_dir),
                 local_dir_use_symlinks=False,
+                **_stop_at_exit(snapshot_download),
             )
             self.logger.info("HF model downloaded")
 

py/lora_import.py

@@ -44,8 +44,10 @@ from .lora_manager_layout import (
 )
 
 try:
+    from ..utils import exit_signal
     from ..utils.hashing import generate_prompt_hash
 except ImportError:  # pragma: no cover - flat test imports
+    from utils import exit_signal
     from utils.hashing import generate_prompt_hash
 
 LORA_CATEGORY = "lora-manager"
@@ -189,14 +191,15 @@ def stage(
     downloads its CivitAI example images. A sidecar that cannot be read, or
     whose record fails to build, is skipped by its name, as are the
     centralized sidecars discovery could not parse. ``cancelled()`` is asked
-    before each sidecar (raises ImportCancelled); ``on_record(index,
+    before each sidecar, and so is the exit signal (#376: the legacy scan's
+    ``cancelled`` does not read it); either raises ImportCancelled. ``on_record(index,
     record_or_None, staged)`` follows each one.
     """
     sidecars = discovery.sidecars
     staged = StagedImport(total=len(sidecars))
     staged.skipped.extend(discovery.unreadable)
     for index, sidecar in enumerate(sidecars):
-        if cancelled():
+        if cancelled() or exit_signal.requested():
             raise ImportCancelled()
         record = None
         path = sidecar.metadata_path

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/unseen-restore.mjs

@@ -0,0 +1,54 @@
+// A restore the open workspace page has not noticed yet (#413).
+//
+// Specs that restore through the API while a page is open check what the page does with a restore it did not see:
+// its next write is refused as "Prompt storage was restored since this dialog opened". But the page's Activity store
+// reads the job list every 15 s (activity-store.mjs POLL_MS), and a read the server answers while the restore job runs
+// makes the page follow it: when it finishes, the page is shown again from the restored data (S9, index.mjs
+// onRestoreFinished), which closes any open dialog. On a slow runner a restore can take longer than the time left to
+// the next read (CI run 40646112668: the restore took 16 s), so whether the page saw it depended on timing.
+//
+// `armUnseenRestores(page)`, before the page loads, passes every job-list read of the page through one route, so it
+// knows which reads the server has not answered yet. `unseenRestore(page, restoring)` then closes the gate, waits
+// for those reads' answers (sent before the restore began, they cannot list it), runs `restoring()`, and only after
+// it has answered lets the held reads go on: they read the restore as already ended, as a tab that read its job list
+// just before and just after the restore does. A restore that is never seen running never reloads the page.
+const JOBS = '**/prompt_manager/jobs';
+const gates = new WeakMap();
+
+/** Route `page`'s job-list reads through the restore gate. Call before the page's first navigation. */
+export async function armUnseenRestores(page) {
+    const gate = { closed: null, inFlight: 0, idle: [] };
+    gates.set(page, gate);
+    await page.route(JOBS, async route => {
+        if (gate.closed) {
+            await gate.closed;
+            return route.fallback();
+        }
+        gate.inFlight += 1;
+        try {
+            const response = await route.fetch();
+            await route.fulfill({ response }).catch(() => {});
+        } catch {
+            await route.abort().catch(() => {});
+        } finally {
+            gate.inFlight -= 1;
+            if (gate.inFlight === 0) for (const resolve of gate.idle.splice(0)) resolve();
+        }
+    });
+}
+
+/** Run `restoring()` (an API restore) unseen by `page`'s Activity store; resolves with what it resolves with. */
+export async function unseenRestore(page, restoring) {
+    const gate = gates.get(page);
+    if (!gate) throw new Error('armUnseenRestores(page) must run before the page loads');
+    let open;
+    gate.closed = new Promise(resolve => { open = resolve; });
+    try {
+        // Reads already on their way were sent before the restore: their answers cannot list it.
+        if (gate.inFlight > 0) await new Promise(resolve => gate.idle.push(resolve));
+        return await restoring();
+    } finally {
+        gate.closed = null;
+        open();
+    }
+}

tests/browser/workspace-history.spec.mjs

@@ -1,5 +1,6 @@
 import {startCoverage,saveCoverage} from '../../tools/vendor/browser-coverage.mjs';
 import {test,expect} from './fixtures.mjs';
+import {armUnseenRestores,unseenRestore} from './unseen-restore.mjs';
 import {readFile,writeFile} from 'node:fs/promises';
 import {createHash} from 'node:crypto';
 
@@ -16,13 +17,13 @@ const marker=(page,id,anchor='10')=>recorded(page).locator(`button[data-run-id="
 const exact=async locator=>locator.evaluate(node=>node.textContent);
 
 test.beforeEach(async({page,request,browserName})=>{
-    await startCoverage(page,browserName);page.historyErrors=[];page.on('pageerror',error=>page.historyErrors.push(error.message));
+    await startCoverage(page,browserName);await armUnseenRestores(page);page.historyErrors=[];page.on('pageerror',error=>page.historyErrors.push(error.message));
     const reset=await request.post('/__test__/reset',{data:{chronological_history:true}});await expect(reset).toBeOK();page.fixture=await reset.json();
 });
 test.afterEach(async({page,browserName})=>{await saveCoverage(page,browserName);expect(page.historyErrors).toEqual([]);});
 
 const command=async(request,data)=>{const response=await request.post('/__test__/history/commands',{data});await expect(response).toBeOK();return response.json();};
-const restoreSnapshot=async(request,bytes)=>{const response=await request.post('/prompt_manager/restore',{multipart:{database_file:{name:'history.db',mimeType:'application/octet-stream',buffer:bytes}}});await expect(response).toBeOK();expect((await response.json()).success).toBe(true);};
+const restoreSnapshot=async(page,request,bytes)=>{const response=await unseenRestore(page,()=>request.post('/prompt_manager/restore',{multipart:{database_file:{name:'history.db',mimeType:'application/octet-stream',buffer:bytes}}}));await expect(response).toBeOK();expect((await response.json()).success).toBe(true);};
 const historyRead=async(request,id)=>{const response=await request.get(`/prompt_manager/library/${id}/history`);await expect(response).toBeOK();return response.json();};
 const frozenContexts=contexts=>contexts.map(({navigation,positive_prompt_id,negative_prompt_id,pair_id,...frozen})=>frozen);
 const immutableReceipts=items=>items.map(({current_candidate,...receipt})=>receipt);
@@ -62,7 +63,7 @@ for(const boundary of ['window','detail','candidate','inspection','media']) test
     await command(request,{action:'arm',boundary});const pending=request.get(target);
     try {
         await expect.poll(async()=> (await(await request.get('/__test__/history/checkpoint')).json()).entered).toBe(true);
-        await restoreSnapshot(request,bytes);
+        await restoreSnapshot(page,request,bytes);
     } finally {await command(request,{action:'release'});}
     const retired=await pending;expect(retired.status()).toBe(409);expect((await retired.json()).code).toBe('history-changed');
     for(const url of [`/prompt_manager/library/${f.detail_id}/history?limit=1&cursor=${encodeURIComponent(firstWindow.next_cursor)}`,`${base}?epoch=${epoch}`,`${base}/outputs?epoch=${epoch}`,`${base}/outputs?epoch=${epoch}&cursor=${encodeURIComponent(outputs.next_cursor)}`,media.url,media.thumbnail_url,media.inspection_url]) {
@@ -130,7 +131,7 @@ test('archive delete merge and prune preserve frozen evidence without repairing
 
 test('actual restore with reused numeric owner rejects stale history and offers explicit saved recovery',async({page,request})=>{
     const f=page.fixture;await open(page,f.detail_id);const original=await exact(saved(page).getByRole('region',{name:'Positive prompt',exact:true}).locator('p.prompt-text'));
-    const source=await request.get('/__test__/history/reused-snapshot');await expect(source).toBeOK();await restoreSnapshot(request,await source.body());
+    const source=await request.get('/__test__/history/reused-snapshot');await expect(source).toBeOK();await restoreSnapshot(page,request,await source.body());
     await marker(page,f.runs.shared).click();await expect(dialog(page)).toContainText('Recorded evidence changed');await page.keyboard.press('Escape');
     await expect(page.locator('.library-parent-status')).toContainText('changed');expect(await exact(saved(page).getByRole('region',{name:'Positive prompt',exact:true}).locator('p.prompt-text'))).toBe(original);
     await expect(recorded(page).getByRole('button',{name:'Reload recorded history',exact:true})).toBeVisible();
@@ -142,12 +143,12 @@ for(const outcome of ['success','error'])test(`delayed actual old browser ${outc
     const f=page.fixture;await open(page,f.detail_id);const snapshot=await request.get('/__test__/history/snapshot');const bytes=await snapshot.body();
     let entered,release;const started=new Promise(resolve=>{entered=resolve;}),held=new Promise(resolve=>{release=resolve;});
     await page.route(new RegExp(`/history/runs/${f.runs.shared}\\?`),async route=>{
-        if(outcome==='error')await restoreSnapshot(request,bytes);
+        if(outcome==='error')await restoreSnapshot(page,request,bytes);
         const response=await route.fetch();expect(response.status()).toBe(outcome==='success'?200:409);entered();await held;
         await route.fulfill({response}).catch(()=>{});
     });
     try {
-        await marker(page,f.runs.shared).click();await started;if(outcome==='success')await restoreSnapshot(request,bytes);
+        await marker(page,f.runs.shared).click();await started;if(outcome==='success')await restoreSnapshot(page,request,bytes);
         await page.evaluate(()=>{location.hash='#/overview';});await expect(page.locator('#workspace-main h1')).toHaveText('Overview');release();
         await page.unroute(new RegExp(`/history/runs/${f.runs.shared}\\?`));await open(page,f.detail_id);
         await expect(recorded(page).locator('button[data-run-id]')).toHaveCount(0);
@@ -258,7 +259,7 @@ test('inspection first discovers a real restore and immediately retires every su
                 if(changed)window.recordedRecoveryStates.push({receipts:run.querySelectorAll('[data-receipt]').length,active:[...run.querySelectorAll('[data-send-role],[data-copy-role],[data-view-receipt]')].filter(node=>!node.disabled&&node.checkVisibility()).length});
             });observer.observe(run,{subtree:true,childList:true,attributes:true,characterData:true});
         });
-        await restoreSnapshot(request,await snapshot.body());
+        await restoreSnapshot(page,request,await snapshot.body());
     } finally {await command(request,{action:'release'});}
     const response=await inspectionRead;expect(response.status()).toBe(409);expect((await response.json()).code).toBe('history-changed');
     await expect(dialog(page).locator('.recorded-run-status')).toContainText('Recorded evidence changed');

tests/browser/workspace-library-editor.spec.mjs

@@ -1,6 +1,7 @@
 import {writeFile} from 'node:fs/promises';
 import {startCoverage,saveCoverage} from '../../tools/vendor/browser-coverage.mjs';
 import {test,expect} from './fixtures.mjs';
+import {armUnseenRestores,unseenRestore} from './unseen-restore.mjs';
 import {captureCanonicalState} from './history-canonical-comparison.mjs';
 import {inventoryOf,mapConcept,diffInventory} from './library-editor-canonical-comparison.mjs';
 
@@ -13,7 +14,7 @@ const RATINGS=[['5','5 / 5'],['4','4 / 5'],['3','3 / 5'],['2','2 / 5'],['1','1 /
 const editorDialog=page=>page.locator('dialog.library-editor');
 
 test.beforeEach(async({page,request,browserName})=>{
-    await startCoverage(page,browserName);
+    await startCoverage(page,browserName);await armUnseenRestores(page);
     const reset=await request.post('/__test__/library-editor/reset',{data:{}});await expect(reset).toBeOK();page.fixture=await reset.json();
     page.editorErrors=[];page.on('pageerror',error=>page.editorErrors.push(error.message));
     page.writes=[];page.on('request',item=>{if(item.method()!=='GET')page.writes.push({method:item.method(),url:new URL(item.url()).pathname,body:item.postDataJSON()});});
@@ -260,7 +261,8 @@ test('recovery states render inline and never replay a write',async({page,reques
     page.writes.length=0;
     dialog=await openEditor(page,{mode:'new'});await ready(dialog);
     const snapshot=await request.get('/__test__/library-editor/snapshot');await expect(snapshot).toBeOK();
-    const restore=await request.post('/prompt_manager/restore',{multipart:{database_file:{name:'editor.db',mimeType:'application/octet-stream',buffer:await snapshot.body()}}});
+    const buffer=await snapshot.body();
+    const restore=await unseenRestore(page,()=>request.post('/prompt_manager/restore',{multipart:{database_file:{name:'editor.db',mimeType:'application/octet-stream',buffer}}}));
     await expect(restore).toBeOK();page.writes.length=0;
     await field(dialog,'Title').fill('After restore');await field(dialog,'Positive prompt').fill('Restored storage prompt');await button(dialog,'Save').click();
     await expect(note(dialog)).toHaveText('Prompt storage was restored since this dialog opened. Nothing was saved.');
@@ -588,7 +590,7 @@ test('saved refresh failure retries only reads',async({page})=>{
 
 test('restored storage after a confirmed save or before an archive offers an explicit Library reload',async({page,request})=>{
     const f=page.fixture;const snapshot=await(await request.get('/__test__/library-editor/snapshot')).body();
-    const restore=()=>request.post('/prompt_manager/restore',{multipart:{database_file:{name:'editor.db',mimeType:'application/octet-stream',buffer:snapshot}}});
+    const restore=()=>unseenRestore(page,()=>request.post('/prompt_manager/restore',{multipart:{database_file:{name:'editor.db',mimeType:'application/octet-stream',buffer:snapshot}}}));
     await openHistory(page,f.positive_id);
     await page.route(`**/prompt_manager/library/${f.positive_id}`,async route=>{
         if(route.request().method()!=='PATCH')return route.continue();
@@ -615,7 +617,7 @@ test('restored storage after a confirmed save or before an archive offers an exp
 const DISCARDS='Leaving this dialog discards your entries.';
 test('restored storage before an Edit or Add negative save offers only the Library reload exit',async({page,request})=>{
     const f=page.fixture;const snapshot=await(await request.get('/__test__/library-editor/snapshot')).body();
-    const restore=()=>request.post('/prompt_manager/restore',{multipart:{database_file:{name:'editor.db',mimeType:'application/octet-stream',buffer:snapshot}}});
+    const restore=()=>unseenRestore(page,()=>request.post('/prompt_manager/restore',{multipart:{database_file:{name:'editor.db',mimeType:'application/octet-stream',buffer:snapshot}}}));
     await openHistory(page,f.positive_id);
     await page.getByRole('button',{name:'Edit metadata',exact:true}).click();let dialog=editorDialog(page);await ready(dialog);
     await expect(await restore()).toBeOK();

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-diagnostics.spec.mjs

@@ -516,7 +516,7 @@ test('a rollover after the review deletes nothing and Activity offers Review aga
 test('Clear all logs states both counts, confirms first and reads everything again', async ({ page, request }) => {
     await seedLogs(request, LOG_FOLDER);
     const truncates = [];
-    page.on('request', item => { if (item.url().endsWith('/prompt_manager/logs/truncate')) truncates.push(item.url()); });
+    page.on('request', item => { if (item.url().endsWith('/prompt_manager/logs/truncate')) truncates.push(item.postDataJSON()); });
     await openDiagnostics(page);
     await expect(lines(page)).toContainText('Database backup verified');
     const details = await openFiles(page);
@@ -535,6 +535,10 @@ test('Clear all logs states both counts, confirms first and reads everything aga
     await dialog(page).getByRole('button', { name: 'Clear', exact: true }).click();
     await expect(toast(page)).toHaveText('Logs cleared · 3 rotated copies deleted.');
     expect(truncates).toHaveLength(1);
+    // #389: Clear sends back exactly the files the dialog was built from, each with its identity string.
+    expect(truncates[0].expected.map(row => row.name).sort()).toEqual(['prompt_manager.log', 'prompt_manager.log.1',
+        'prompt_manager.log.2', 'prompt_manager.log.3']);
+    for (const row of truncates[0].expected) expect(row.identity).toMatch(/^\d+:\d+:\d+$/);
     // Read again: the viewer, the file line and the list.
     await expect(lines(page)).not.toContainText('Database backup verified');
     await expect(details.locator(':scope > p.small').first()).toHaveText(/rotated copies: 0$/);
@@ -564,6 +568,23 @@ test('Clear all logs reads the files again before it states what it deletes', as
     expect(await names(request)).toEqual(['prompt_manager.log', 'prompt_manager.log.1', 'prompt_manager.log.2', 'prompt_manager.log.3']);
 });
 
+test('a rollover while Clear all logs is open deletes nothing and says why (#389)', async ({ page, request }) => {
+    await seedLogs(request, { rotated: [1024, 1024], active: 100 });
+    await openDiagnostics(page);
+    await openFiles(page);
+    await operation(page, 'Clear all logs').getByRole('button', { name: 'Clear all logs…', exact: true }).click();
+    await expect(dialog(page).locator('#modal-body > p')).toHaveText(/^Empties prompt_manager\.log, deletes 2 rotated copies /);
+    // The log rolls over after the confirmation was built: the copies it listed are not the ones there now.
+    await expect(await request.post('/__test__/logs/rotate')).toBeOK();
+    const rolled = await names(request);
+    const answer = page.waitForResponse(response => response.url().endsWith('/prompt_manager/logs/truncate'));
+    await dialog(page).getByRole('button', { name: 'Clear', exact: true }).click();
+    expect((await answer).status()).toBe(409);
+    await expect(toast(page)).toHaveText('The log files changed since you confirmed; nothing was cleared. '
+        + 'Read the list again and clear again.');
+    expect(await names(request)).toEqual(rolled);
+});
+
 test('each listed file downloads by its returned name and a link has no Download', async ({ page, request }) => {
     await seedLogs(request, LOG_FOLDER);
     await openDiagnostics(page);

tests/browser/workspace-vocabulary.spec.mjs

@@ -1,6 +1,7 @@
 import {startCoverage,saveCoverage} from '../../tools/vendor/browser-coverage.mjs';
 import {writeFile} from 'node:fs/promises';
 import {test,expect} from './fixtures.mjs';
+import {armUnseenRestores,unseenRestore} from './unseen-restore.mjs';
 import {captureCanonicalState} from './history-canonical-comparison.mjs';
 import {pageInventoryOf,inventoryOf,diffInventory,mapVocabularyConcept,mapVocabularyDialog} from './vocabulary-canonical-comparison.mjs';
 
@@ -40,7 +41,7 @@ async function expectLibrary(page,{tags='',category='',count}) {
 
 let manifest;
 test.beforeEach(async({page,request,browserName})=>{
-    await startCoverage(page,browserName);manifest=await seed(request);
+    await startCoverage(page,browserName);await armUnseenRestores(page);manifest=await seed(request);
     page.vocabularyErrors=[];page.on('pageerror',error=>page.vocabularyErrors.push(error.message));
 });
 test.afterEach(async({page,browserName})=>{await saveCoverage(page,browserName);expect(page.vocabularyErrors).toEqual([]);});
@@ -264,7 +265,8 @@ const toast=page=>page.locator('#workspace-toast');
 const actionNames=scope=>scope.locator('.dialog-actions button').evaluateAll(nodes=>nodes.map(node=>node.textContent.trim()));
 const snapshot=async request=>{const response=await request.get('/__test__/vocabulary/snapshot');await expect(response).toBeOK();return response.json();};
 const backup=async request=>{const response=await request.get('/__test__/vocabulary/backup');await expect(response).toBeOK();return response.body();};
-const restore=async(request,buffer)=>expect(await request.post('/prompt_manager/restore',{multipart:{database_file:{name:'vocabulary.db',mimeType:'application/octet-stream',buffer}}})).toBeOK();
+// A restore the open page has not seen, so its dialog stays to report it (#413).
+const restore=async(page,request,buffer)=>expect(await unseenRestore(page,()=>request.post('/prompt_manager/restore',{multipart:{database_file:{name:'vocabulary.db',mimeType:'application/octet-stream',buffer}}}))).toBeOK();
 const tagNames=rows=>rows.tags.map(([,name])=>name);
 function trackWrites(page) {
     const writes=[];page.on('request',item=>{if(item.method()!=='GET')writes.push({method:item.method(),url:new URL(item.url()).pathname,body:item.postDataJSON()});});return writes;
@@ -482,7 +484,7 @@ test('conflict and restore states keep the dialog and write nothing',async({page
 
     const before=await snapshot(request);
     box=await openDialog(page,main(page).getByRole('button',{name:'+ Create tag',exact:true}),'Create tag');
-    await restore(request,saved);
+    await restore(page,request,saved);
     await field(box,'Tag name').fill('sunrise');await button(box,'Apply').click();
     await expect(note(box)).toHaveText('Prompt storage was restored since this dialog opened. Nothing was changed.');
     await expect(button(box,'Apply')).toBeDisabled();
@@ -622,7 +624,7 @@ test('delete confirmation read failure, missing tag and restore keep Delete disa
     await expect(button(box,'Delete')).toBeDisabled();await expect(tagRow(page,'calm').locator('td:nth-child(2)')).toHaveText('2');
     await button(box,'Cancel').click();
 
-    box=await openRename(page,'mist');await restore(request,saved);
+    box=await openRename(page,'mist');await restore(page,request,saved);
     await button(box,'Delete tag…').click();
     await expect(note(box)).toHaveText('Prompt storage was restored since this dialog opened. Nothing was changed.');
     await expect(button(box,'Delete')).toBeDisabled();expect(await actionNames(box)).toEqual(['Reload','Cancel','Delete']);

tests/browser/workspace_diagnostics_fixture.py

@@ -123,6 +123,34 @@ def _touch(path: Path, size: int, moment: datetime.datetime) -> None:
     os.utime(path, (stamp, stamp))
 
 
+def _append_active(manager, log_dir: Path, size: int) -> None:
+    """Grow the active log as this process's handler does (#389).
+
+    A second handle appending to the file the handler writes is what Clear
+    all logs refuses as another program writing to it, so the bytes go
+    through the handler's own stream; with file logging off, to the file.
+    """
+
+    def same(path) -> bool:
+        return os.path.normcase(os.path.realpath(path)) == os.path.normcase(
+            os.path.realpath(log_dir / ACTIVE)
+        )
+
+    for handler in manager._file_handlers():
+        stream = getattr(handler, "stream", None)
+        if stream is None or not same(handler.baseFilename):
+            continue
+        handler.acquire()
+        try:
+            stream.write("x" * size)
+            stream.flush()
+        finally:
+            handler.release()
+        return
+    with open(log_dir / ACTIVE, "ab") as handle:
+        handle.write(b"x" * size)
+
+
 def _bounded(value, name, top=MAX_SEED_BYTES):
     if isinstance(value, bool) or not isinstance(value, int) or not 0 <= value <= top:
         raise web.HTTPBadRequest(text=f"{name} must be an integer 0..{top}")
@@ -169,8 +197,7 @@ def register_diagnostics_commands(app, routes, api, root: Path):
         for index, (time_text, level, message) in enumerate(CONCEPT_RECORDS):
             record = _record(time_text, level, message, index)
             manager.add_to_buffer(record, record.getMessage())
-        with open(log_dir / ACTIVE, "ab") as handle:
-            handle.write(b"x" * CONCEPT_ACTIVE)
+        _append_active(manager, log_dir, CONCEPT_ACTIVE)
         for number, size in enumerate(CONCEPT_ROTATED, start=1):
             moment = ROTATED_TIME - datetime.timedelta(hours=number)
             _touch(log_dir / f"{ACTIVE}.{number}", size, moment)
@@ -194,8 +221,7 @@ def register_diagnostics_commands(app, routes, api, root: Path):
             manager.add_to_buffer(record, record.getMessage())
         active = _bounded(data.get("active", 0), "active")
         if active:
-            with open(log_dir / ACTIVE, "ab") as handle:
-                handle.write(b"x" * active)
+            _append_active(manager, log_dir, active)
         rotated = data.get("rotated", [])
         if not isinstance(rotated, list) or len(rotated) > 20:
             raise web.HTTPBadRequest(text="rotated must be a list of sizes")

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/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/js/workspace-system-diagnostics.test.js

@@ -175,6 +175,28 @@ test('Clear all logs states both counts and its toast says what was done', async
     assert.equal(clearToast(undefined), 'Logs cleared · 0 rotated copies deleted.');
 });
 
+test('Clear all logs sends back exactly the files its confirmation showed (#389)', async () => {
+    const { clearBody, clearFailureText, COPY } = await diagnostics;
+    const files = [
+        { filename: 'prompt_manager.log', kind: 'active', identity: '10:1761000000123456789:9007199254740993' },
+        { filename: 'prompt_manager.log.1', kind: 'rotated', identity: '20:1761000000123456789:42' },
+        { filename: 'prompt_manager.log.4', kind: 'link', identity: '5:1:7' },
+        { filename: 'prompt_manager.log.bak', kind: 'other', identity: '1:1:8' },
+        { filename: 'prompt_manager.log.2', kind: 'rotated' },
+    ];
+    // Identities stay strings: mtime_ns and large inodes are not exact JS numbers.
+    assert.deepEqual(clearBody(files), { expected: [
+        { name: 'prompt_manager.log', identity: '10:1761000000123456789:9007199254740993' },
+        { name: 'prompt_manager.log.1', identity: '20:1761000000123456789:42' },
+    ] });
+    assert.deepEqual(clearBody([]), { expected: [] });
+    assert.deepEqual(clearBody(null), {});
+    const refused = Object.assign(new Error('The log files changed since you confirmed; nothing was cleared.'), { status: 409 });
+    assert.equal(clearFailureText(refused), refused.message);
+    assert.equal(clearFailureText(Object.assign(new Error('boom'), { status: 500 })), COPY.clearFailed);
+    assert.equal(clearFailureText(undefined), COPY.clearFailed);
+});
+
 test('the test-link dialog names the prompt and how Missing-file links removes the record', async () => {
     const { testLinkSentence, promptName } = await diagnostics;
     assert.equal(testLinkSentence({ id: 812, text: 'Misty harbor' }), 'Writes a placeholder image record '

tests/test_api_log_cleanup.py

@@ -415,5 +415,74 @@ class TestApply(LogCleanupTestCase):
         self.assertEqual(job.status_line, "The log cleanup failed.")
 
 
+TRUNCATE = "/prompt_manager/logs/truncate"
+FILES = "/prompt_manager/logs/files"
+
+
+class TestClearAllLogsConfirmed(LogCleanupTestCase):
+    """#389: POST /logs/truncate deletes no more than the confirmation listed."""
+
+    async def confirmed(self):
+        response = await self.client.request("GET", FILES)
+        data = await response.json()
+        return [
+            {"name": row["filename"], "identity": row["identity"]}
+            for row in data["files"]
+            if row["kind"] in ("active", "rotated")
+        ]
+
+    def handler(self):
+        return next(
+            h
+            for h in self.manager.logger.handlers
+            if isinstance(h, logging.FileHandler)
+        )
+
+    async def test_a_rollover_after_the_confirmation_is_refused(self):
+        self.write("prompt_manager.log.1", b"a" * 100)
+        self.write("prompt_manager.log.2", b"b" * 200)
+        expected = await self.confirmed()
+        self.app_logger.info("written after the confirmation")
+        self.handler().doRollover()
+        before = self.snapshot()
+        body = await self.post(TRUNCATE, {"expected": expected}, 409)
+        self.assertEqual(body["success"], False)
+        self.assertEqual(body["code"], "logs-changed")
+        self.assertIn("nothing was cleared", body["error"])
+        self.assertEqual(self.snapshot(), before)
+
+    async def test_the_confirmed_files_are_cleared(self):
+        self.write("prompt_manager.log.1", b"a" * 100)
+        expected = await self.confirmed()
+        body = await self.post(TRUNCATE, {"expected": expected}, 200)
+        self.assertEqual(
+            sorted(body["results"]["truncated"]),
+            ["prompt_manager.log", "prompt_manager.log.1"],
+        )
+        self.assertEqual(self.names(), ["prompt_manager.log"])
+
+    async def test_the_legacy_call_without_a_body_still_clears(self):
+        self.write("prompt_manager.log.1", b"a" * 100)
+        response = await self.client.request("POST", TRUNCATE)
+        self.assertEqual(response.status, 200, await response.text())
+        self.assertEqual(self.names(), ["prompt_manager.log"])
+
+    async def test_a_foreign_writer_is_refused_with_a_clear_message(self):
+        self.write("prompt_manager.log.1", b"a" * 100)
+        with open(self.path("prompt_manager.log"), "ab") as other:
+            other.write(b"line from another ComfyUI\n")
+        body = await self.post(TRUNCATE, {"expected": await self.confirmed()}, 409)
+        self.assertEqual(body["code"], "logs-foreign-writer")
+        self.assertIn("Another program is writing", body["error"])
+        self.assertIn("prompt_manager.log.1", self.names())
+
+    async def test_a_malformed_confirmation_is_400(self):
+        self.write("prompt_manager.log.1", b"a" * 100)
+        for expected in ("all", [{"name": "prompt_manager.log.1", "identity": 5}]):
+            with self.subTest(expected=expected):
+                await self.post(TRUNCATE, {"expected": expected}, 400)
+        self.assertIn("prompt_manager.log.1", self.names())
+
+
 if __name__ == "__main__":
     unittest.main()

tests/test_api_review.py

@@ -1040,8 +1040,8 @@ class TestApplyRefusals(ReviewRoutesTestCase):
         _path, item, detail, observed, cat = await self._opened()
         real = review_capture_module.resolve_prompt_pair
 
-        def reworded(metadata):
-            resolution = real(metadata)
+        def reworded(metadata, **kwargs):
+            resolution = real(metadata, **kwargs)
             return {
                 **resolution,
                 "reason": "a reworded reason from a newer resolver",
@@ -1065,8 +1065,8 @@ class TestApplyRefusals(ReviewRoutesTestCase):
         _path, item, detail, observed, cat = await self._opened()
         real = review_capture_module.resolve_prompt_pair
 
-        def without_cat(metadata):
-            resolution = real(metadata)
+        def without_cat(metadata, **kwargs):
+            resolution = real(metadata, **kwargs)
             kept = [
                 c
                 for c in resolution["candidates"]

tests/test_exit_cancel.py

@@ -0,0 +1,463 @@
+"""Work running at Ctrl+C stops at its next safe point (#376).
+
+ComfyUI catches Ctrl+C in ``main.py`` and returns without closing its event
+loop. Python then joins every executor thread (``concurrent.futures``'s
+``_python_exit``, a ``threading._register_atexit`` callback) before any
+``atexit`` callback runs, so a call still running on the default executor
+used to keep the process alive until it returned. The package registers its
+own ``threading._register_atexit`` hook, which runs before that join: it sets
+the exit signal that ``Job.cancelled`` and ``map_parallel`` read, and kills
+the encoder processes.
+
+The child-process test starts three long calls on the default executor (a
+registry job checking ``job.cancelled`` between files, a ``map_parallel``
+batch and an encoder process that ignores the exit, as ffmpeg does on
+Windows), returns from its main module and must exit far sooner than any of
+them would end.
+"""
+
+import json
+import os
+import subprocess
+import sys
+import tempfile
+import textwrap
+import threading
+import unittest
+from unittest import mock
+
+ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+sys.path.insert(0, ROOT)
+
+try:
+    from utils import exit_signal  # noqa: E402
+except ImportError:  # before #376: only the child-process test can run
+    exit_signal = None
+
+# The parent never waits longer than this for the child (the hang watchdog
+# allows 120 s per test). Each fake job alone would run 300 s or more.
+CHILD_CEILING_SECONDS = 60
+# From the child's return to its last atexit callback: the stopped calls end
+# at their next file, item or chunk (the encoder is killed).
+EXIT_BOUND_SECONDS = 15
+JOB_SECONDS = 600
+# The fake encoder outlives the ceiling, yet a failing run leaks it only
+# briefly (the parent cannot reach the child's own child to kill it).
+ENCODER_SECONDS = 90
+REPORT_PREFIX = "PM_EXIT_REPORT "
+
+CHILD = textwrap.dedent("""
+    import asyncio
+    import atexit
+    import contextlib
+    import importlib
+    import io
+    import json
+    import os
+    import sys
+    import threading
+    import time
+
+    root, tmp = sys.argv[1], sys.argv[2]
+    job_seconds, encoder_seconds = float(sys.argv[3]), int(sys.argv[4])
+    sys.path.insert(0, root)
+    state = {"stopped": {}}
+
+    def report():
+        returned = state.get("returned")
+        print("PM_EXIT_REPORT " + json.dumps({
+            "exit_seconds": None if returned is None else time.monotonic() - returned,
+            "stopped": state["stopped"],
+        }))
+        sys.stdout.flush()
+
+    # Registered first, so it runs after every other atexit callback, and
+    # every atexit callback runs after the executor threads were joined.
+    atexit.register(report)
+
+    from tests.test_init_package import _import_package, fake_server_module
+
+    output = os.path.join(tmp, "output")
+    with contextlib.redirect_stdout(io.StringIO()):
+        package, _ = _import_package(
+            "pm_exit_child",
+            os.path.join(tmp, "prompts.db"),
+            output,
+            fake_server_module(queue_task_done=True),
+        )
+    parallel = importlib.import_module("pm_exit_child.utils.parallel")
+    encode = importlib.import_module("pm_exit_child.py.api.thumbnail_encode")
+    registry = package.api._jobs
+    entered = {name: threading.Event() for name in ("job", "batch", "encoder")}
+
+    def job_work(job):
+        # A job's thread work: one "file" at a time, checking its cancel
+        # between files the way the thumbnail and import jobs do.
+        entered["job"].set()
+        deadline = time.monotonic() + job_seconds
+        while time.monotonic() < deadline:
+            if job.cancelled:
+                state["stopped"]["job"] = True
+                return
+            time.sleep(0.05)
+
+    def batch_item(index):
+        entered["batch"].set()
+        time.sleep(0.5)
+        return index
+
+    def batch_work():
+        try:
+            parallel.map_parallel(batch_item, range(int(job_seconds)), 2)
+        except Exception as error:
+            state["stopped"]["batch"] = type(error).__name__
+
+    def encoder_work():
+        # Stands for ffmpeg on Windows, which runs without a console and
+        # never sees Ctrl+C: only a kill ends it before its timeout.
+        cmd = [sys.executable, "-c", "import time; time.sleep(%d)" % encoder_seconds]
+        try:
+            encode.run_encoder(
+                cmd,
+                timeout=encoder_seconds,
+                track=lambda _process: entered["encoder"].set(),
+            )
+        except Exception as error:
+            state["stopped"]["encoder"] = type(error).__name__
+
+    async def begin():
+        loop = asyncio.get_running_loop()
+
+        async def runner(job):
+            await loop.run_in_executor(None, job_work, job)
+
+        registry.start("exit-test", "Exit test", admission="read", runner=runner)
+        loop.run_in_executor(None, batch_work)
+        loop.run_in_executor(None, encoder_work)
+        for name, event in entered.items():
+            if not await loop.run_in_executor(None, event.wait, 30):
+                sys.exit("not started: " + name)
+
+    loop = asyncio.new_event_loop()
+    asyncio.set_event_loop(loop)
+    loop.run_until_complete(begin())
+    state["returned"] = time.monotonic()
+    # Return with the three calls running and the loop open, as ComfyUI does
+    # after Ctrl+C.
+    """)
+
+
+class TestExitStopsRunningWork(unittest.TestCase):
+    def test_running_executor_work_stops_at_exit(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            os.makedirs(os.path.join(tmp, "output"))
+            script = os.path.join(tmp, "child.py")
+            with open(script, "w", encoding="utf-8") as handle:
+                handle.write(CHILD)
+            env = dict(os.environ)
+            env.pop("PROMPT_MANAGER_CONFIG_PATH", None)
+            env["PROMPT_MANAGER_DATA_DIR"] = tmp
+            env["PYTHONIOENCODING"] = "utf-8"
+            try:
+                completed = subprocess.run(
+                    [
+                        sys.executable,
+                        script,
+                        ROOT,
+                        tmp,
+                        str(JOB_SECONDS),
+                        str(ENCODER_SECONDS),
+                    ],
+                    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.assertLess(report["exit_seconds"], EXIT_BOUND_SECONDS, detail)
+        self.assertEqual(
+            report["stopped"],
+            {"job": True, "batch": "ExitRequested", "encoder": "EncoderClosed"},
+            detail,
+        )
+        for text in ("Traceback", "Background work not stopped in time"):
+            self.assertNotIn(text, completed.stderr, detail)
+
+
+class _Isolated(unittest.TestCase):
+    """Each test starts with the signal clear and no hook installed."""
+
+    def setUp(self):
+        exit_signal.reset()
+        self.addCleanup(exit_signal.reset)
+
+
+class TestInstall(_Isolated):
+    def test_the_hook_is_registered_once_before_the_executor_join(self):
+        calls = []
+        with mock.patch.object(
+            exit_signal.threading,
+            "_register_atexit",
+            side_effect=lambda func: calls.append(func),
+            create=True,
+        ):
+            self.assertTrue(exit_signal.install())
+            self.assertTrue(exit_signal.install())
+        self.assertEqual(len(calls), 1)
+        self.assertFalse(exit_signal.requested())
+        calls[0]()
+        self.assertTrue(exit_signal.requested())
+
+    def test_a_missing_internal_hook_degrades_to_the_old_exit(self):
+        with mock.patch.object(exit_signal, "threading") as fake:
+            del fake._register_atexit
+            self.assertFalse(exit_signal.install())
+        self.assertFalse(exit_signal.requested())
+
+    def test_a_refused_registration_degrades_to_the_old_exit(self):
+        refuse = RuntimeError("can't register atexit after shutdown")
+        with mock.patch.object(
+            exit_signal.threading, "_register_atexit", side_effect=refuse, create=True
+        ):
+            self.assertFalse(exit_signal.install())
+
+    def test_a_failing_exit_action_never_escapes_the_hook(self):
+        # An exception out of a threading atexit callback would abort
+        # threading._shutdown and skip the executor join itself.
+        calls = []
+
+        def boom():
+            raise KeyboardInterrupt  # a second Ctrl+C
+
+        with mock.patch.object(
+            exit_signal.threading,
+            "_register_atexit",
+            side_effect=lambda func: calls.append(func),
+            create=True,
+        ):
+            exit_signal.install(boom)
+        with self.assertLogs(exit_signal.logger, "ERROR"):
+            calls[0]()
+        self.assertTrue(exit_signal.requested())
+
+
+class TestCheckpoints(_Isolated):
+    def test_checked_runs_until_the_signal_then_raises(self):
+        seen = []
+        record = exit_signal.checked(seen.append)
+        record(1)
+        exit_signal.request()
+        with self.assertRaises(exit_signal.ExitRequested):
+            record(2)
+        self.assertEqual(seen, [1])
+
+    def test_map_parallel_starts_no_item_after_the_signal(self):
+        from utils.parallel import map_parallel
+
+        started = []
+        gate = threading.Event()
+
+        def item(index):
+            started.append(index)
+            if index == 0:
+                exit_signal.request()
+                gate.set()
+            return index
+
+        with self.assertRaises(exit_signal.ExitRequested):
+            map_parallel(item, range(50), 1)
+        self.assertTrue(gate.is_set())
+        self.assertEqual(started, [0])
+
+    def test_job_cancelled_follows_the_signal_and_its_own_cancel(self):
+        from py.api.jobs import Job
+
+        job = Job("scan", "Scan", admission="records")
+        self.assertFalse(job.cancelled)
+        job.cancelled = True
+        self.assertTrue(job.cancelled)
+        other = Job("scan", "Scan", admission="records", cancellable=False)
+        exit_signal.request()
+        self.assertTrue(other.cancelled)
+
+    def test_the_scan_walk_stops_between_directories(self):
+        from py.api.admin import _collect_media_files
+
+        with tempfile.TemporaryDirectory() as tmp:
+            for name in ("a", "b"):
+                os.makedirs(os.path.join(tmp, name))
+                with open(os.path.join(tmp, name, "x.png"), "wb") as image:
+                    image.write(b"x")
+            self.assertEqual(len(_collect_media_files([tmp], [".png"])), 2)
+            exit_signal.request()
+            with self.assertRaises(exit_signal.ExitRequested):
+                _collect_media_files([tmp], [".png"])
+
+
+class TestLoraImportStopsAtExit(_Isolated):
+    def test_staging_stops_before_the_next_sidecar_whatever_the_caller_checks(self):
+        import types
+
+        from py import lora_import
+
+        discovery = types.SimpleNamespace(
+            sidecars=[object(), object()], unreadable=[], layout=None
+        )
+        exit_signal.request()
+        with tempfile.TemporaryDirectory() as tmp:
+            with self.assertRaises(lora_import.ImportCancelled):
+                lora_import.stage(
+                    discovery,
+                    tmp,
+                    media_extensions=frozenset(),
+                    cancelled=lambda: False,  # the legacy scan's own check
+                )
+
+
+class _FakeTqdm:
+    """huggingface_hub.utils.tqdm as far as a download uses it."""
+
+    def __init__(self, *args, **kwargs):
+        self.n = 0
+
+    def update(self, n=1):
+        self.n += n
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *exc):
+        return False
+
+
+def _fake_hub(chunks_before_exit):
+    """A ``huggingface_hub`` whose download writes chunks to a temp file and,
+    like huggingface_hub 1.x, removes it on any failure before the rename."""
+    import types
+
+    def hf_hub_download(
+        repo_id, filename, local_dir=None, local_dir_use_symlinks=None, tqdm_class=None
+    ):
+        final = os.path.join(local_dir, filename)
+        partial = final + ".abc.incomplete"
+        bar = (tqdm_class or _FakeTqdm)(total=10)
+        try:
+            with open(partial, "wb") as handle:
+                for index in range(10):
+                    if index == chunks_before_exit:
+                        exit_signal.request()
+                    bar.update(1)
+                    handle.write(b"x")
+            os.replace(partial, final)
+        finally:
+            if os.path.exists(partial):
+                os.unlink(partial)
+        return final
+
+    hub = types.ModuleType("huggingface_hub")
+    hub.hf_hub_download = hf_hub_download
+    hub.snapshot_download = lambda **kwargs: None
+    utils = types.ModuleType("huggingface_hub.utils")
+    utils.tqdm = _FakeTqdm
+    hub.utils = utils
+    return {"huggingface_hub": hub, "huggingface_hub.utils": utils}
+
+
+class TestModelDownloadStopsAtExit(_Isolated):
+    def test_a_download_stops_between_chunks_and_leaves_no_file(self):
+        from py.autotag import AutoTagService
+
+        with tempfile.TemporaryDirectory() as tmp:
+            service = AutoTagService(models_dir=tmp)
+            with mock.patch.dict(sys.modules, _fake_hub(chunks_before_exit=3)):
+                with self.assertLogs(service.logger, "INFO") as logs:
+                    self.assertFalse(service.download_model("wd14-vit"))
+            folder = os.path.join(tmp, "WD14-ViT")
+            self.assertEqual(os.listdir(folder), [])
+        self.assertTrue(any("exiting" in line for line in logs.output), logs.output)
+        self.assertFalse(any("Download failed" in line for line in logs.output))
+
+    def test_a_download_without_the_signal_finishes(self):
+        from py.autotag import AutoTagService
+
+        with tempfile.TemporaryDirectory() as tmp:
+            service = AutoTagService(models_dir=tmp)
+            with mock.patch.dict(sys.modules, _fake_hub(chunks_before_exit=None)):
+                self.assertTrue(service.download_model("wd14-vit"))
+            folder = os.path.join(tmp, "WD14-ViT")
+            self.assertEqual(
+                sorted(os.listdir(folder)), ["model.onnx", "selected_tags.csv"]
+            )
+
+    def test_xet_transfers_are_aborted_the_way_hub_aborts_on_ctrl_c(self):
+        import types
+
+        from py.autotag import abort_hub_downloads
+
+        aborted = []
+        xet = types.ModuleType("huggingface_hub.utils._xet")
+        xet.abort_xet_session = lambda: aborted.append(True)
+        hub = _fake_hub(chunks_before_exit=None)
+        hub["huggingface_hub.utils._xet"] = xet
+        hub["huggingface_hub.utils"]._xet = xet
+        with mock.patch.dict(sys.modules, hub):
+            abort_hub_downloads()
+        self.assertEqual(aborted, [True])
+        with mock.patch.dict(sys.modules):
+            sys.modules.pop("huggingface_hub", None)
+            abort_hub_downloads()  # never imported: nothing to abort
+        self.assertEqual(aborted, [True])
+
+    def test_an_aborted_xet_download_is_logged_as_stopped(self):
+        from py.autotag import AutoTagService
+
+        hub = _fake_hub(chunks_before_exit=None)
+
+        def aborted(*args, **kwargs):
+            exit_signal.request()  # the hook ran, then hub raised
+            raise RuntimeError("xet session aborted")
+
+        hub["huggingface_hub"].hf_hub_download = aborted
+        with tempfile.TemporaryDirectory() as tmp:
+            service = AutoTagService(models_dir=tmp)
+            with mock.patch.dict(sys.modules, hub):
+                with self.assertLogs(service.logger, "INFO") as logs:
+                    self.assertFalse(service.download_model("wd14-vit"))
+        self.assertTrue(any("exiting" in line for line in logs.output), logs.output)
+        self.assertFalse(any("Download failed" in line for line in logs.output))
+
+    def test_an_old_hub_without_tqdm_class_downloads_as_before(self):
+        from py.autotag import AutoTagService
+
+        hub = _fake_hub(chunks_before_exit=None)
+        full = hub["huggingface_hub"].hf_hub_download
+
+        def old_download(
+            repo_id, filename, local_dir=None, local_dir_use_symlinks=None
+        ):
+            return full(repo_id, filename, local_dir, local_dir_use_symlinks)
+
+        hub["huggingface_hub"].hf_hub_download = old_download
+        with tempfile.TemporaryDirectory() as tmp:
+            service = AutoTagService(models_dir=tmp)
+            with mock.patch.dict(sys.modules, hub):
+                self.assertTrue(service.download_model("wd14-vit"))
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_review_inspection.py

@@ -298,19 +298,21 @@ class TestQwenEditBatchIsQueued(InspectionTestCase):
 
 class TestDefaultProfileStillDecides(InspectionTestCase):
     async def test_ambiguous_file_with_texts_keeps_the_default_profile(self):
-        """A default-ambiguous file with a choice never runs the inspection
-        profile, even when that profile would offer more candidates."""
+        """A default-ambiguous file with a choice is recorded from the
+        default profile, even when the inspection profile would offer more
+        candidates. That profile only checks the saver anchor (#390): no
+        saver names this file, so it anchors nothing."""
         self._roots()
         graph = {
             **_prefixed(qwen_edit_graph("make it night"), "q"),
             **unlinkable_graph("cat", "dog"),
         }
-        self._write(graph=graph)
+        self._write("unsaved_00001_.png", graph=graph)
 
         with patch.object(
             review_capture,
-            "resolve_inspection",
-            side_effect=AssertionError("inspection profile ran"),
+            "inspection_choice",
+            side_effect=AssertionError("inspection choices were asked"),
         ):
             await self._scan()
 
@@ -321,17 +323,24 @@ class TestDefaultProfileStillDecides(InspectionTestCase):
         self.assertEqual(item["resolver_version"], review_capture.RESOLVER_VERSION)
         self.assertEqual(_texts(evidence), ["cat", "dog"])
 
-    async def test_apply_of_a_default_item_never_runs_the_inspection_profile(self):
+    async def test_apply_of_a_default_item_inspects_only_for_the_anchor(
+        self,
+    ):
         self._roots()
         write_png(self.output_dir / "two.png", prompt=unlinkable_graph("cat", "dog"))
         await self._scan()
-        with patch.object(
-            review_capture,
-            "resolve_inspection",
-            side_effect=AssertionError("inspection profile ran"),
+        with (
+            _counted(review_capture, "resolve_inspection") as calls,
+            patch.object(
+                review_capture,
+                "inspection_choice",
+                side_effect=AssertionError("inspection choices were asked"),
+            ),
         ):
             _item, (status, data) = await self._apply("cat")
         self.assertEqual(status, 200, data)
+        # The saver anchor check (#390): with the file's name, nothing matches.
+        self.assertEqual([c[1] for c in calls], ["two.png"])
         self.assertEqual(self._linked_text(self.output_dir / "two.png"), "cat")
 
 
@@ -369,15 +378,16 @@ class TestApplyInspectionCandidate(InspectionTestCase):
         calls = []
         real = review_capture.resolve_inspection
 
-        def spy(metadata):
-            calls.append(metadata)
-            return real(metadata)
+        def spy(metadata, file_name=None):
+            calls.append((metadata, file_name))
+            return real(metadata, file_name)
 
         with patch.object(review_capture, "resolve_inspection", side_effect=spy):
             _item, (status, data) = await self._apply("make it night")
 
         self.assertEqual(status, 200, data)
         self.assertEqual(len(calls), 1)
+        self.assertEqual(calls[0][1], NAME)  # anchored on its saver (#390)
 
     async def test_table_change_keeps_the_item_applicable(self):
         """A table change for a class the graph uses moves the item's
@@ -853,8 +863,8 @@ class TestInspectionResolverVersion(InspectionTestCase):
         version = _version()
         real = review_capture.resolve_inspection
 
-        def newer(metadata):
-            resolution = real(metadata)
+        def newer(metadata, file_name=None):
+            resolution = real(metadata, file_name)
             return {
                 **resolution,
                 "candidates": [
@@ -899,10 +909,10 @@ class TestScanContainment(InspectionTestCase):
         self._write("b.png", graph=qwen_edit_graph("snow", "rain"))
         real = review_capture.resolve_inspection
 
-        def fails_for_a(metadata):
+        def fails_for_a(metadata, file_name=None):
             if "make it night" in json.dumps(metadata):
                 raise RuntimeError("inspection blew up")
-            return real(metadata)
+            return real(metadata, file_name)
 
         with (
             patch.object(review_capture, "resolve_inspection", side_effect=fails_for_a),

tests/test_review_saver_anchor.py

@@ -0,0 +1,333 @@
+"""The scan and Needs review anchor a file on the saver that wrote it (#390).
+
+The workspace inspector (since #383) passes the file's name to the resolver,
+which anchors the file on the one saver whose ``filename_prefix`` names it
+(``_GraphResolver.saver_stages``). The scan and Needs review use the same
+rule. Maintainer ruling: a file never falls through both the link and the
+queue, so when only that anchor resolves a file the default profile and the
+parser floor do not link, it is queued with the anchor's pair as its one
+choice (never linked automatically). No match, several matches or an
+unfinished walk anchor nothing, so such a file is queued as before. Apply
+re-resolves with the file's name, so a choice the anchor contradicts is
+refused as stale evidence.
+"""
+
+import json
+from unittest.mock import patch
+
+try:
+    from tests.test_review import unlinkable_graph, write_png
+    from tests.test_review_inspection import (
+        PROMPTS,
+        _counted,
+        InspectionTestCase,
+        qwen_edit_graph,
+    )
+except ImportError:  # discovered with tests/ as the top-level directory
+    from test_review import unlinkable_graph, write_png
+    from test_review_inspection import (
+        PROMPTS,
+        _counted,
+        InspectionTestCase,
+        qwen_edit_graph,
+    )
+
+from utils import review_capture
+
+
+def branch_graph(*prompts, prefixes=None):
+    """A Qwen-edit batch whose every SaveImage has its own literal prefix."""
+    graph = qwen_edit_graph(*prompts)
+    savers = [n for n in graph.values() if n["class_type"] == "SaveImage"]
+    names = prefixes or [f"branch_{index}" for index in range(len(savers))]
+    for saver, prefix in zip(savers, names):
+        saver["inputs"]["filename_prefix"] = prefix
+    return graph
+
+
+def saved_unlinkable_graph(*positives):
+    """``unlinkable_graph`` whose branches save under their own prefix.
+
+    The default profile finds it ambiguous with a choice and the scan's
+    browser-parser floor links nothing, so it is queued from the default
+    profile unless the saver that wrote the file anchors its branch.
+    """
+    graph = unlinkable_graph(*positives)
+    samplers = [
+        node_id
+        for node_id, node in graph.items()
+        if node["class_type"] == "KSampler" and node_id != "3"
+    ]
+    for index, sampler in enumerate(samplers):
+        graph[f"{sampler}1"] = {
+            "class_type": "VAEDecode",
+            "inputs": {"samples": [sampler, 0], "vae": ["1", 2]},
+        }
+        graph[f"{sampler}2"] = {
+            "class_type": "SaveImage",
+            "inputs": {
+                "images": [f"{sampler}1", 0],
+                "filename_prefix": f"side_{index}",
+            },
+        }
+    return graph
+
+
+def _without_name(real):
+    """``resolve_inspection`` as it was before #390: the file name ignored."""
+
+    def unnamed(metadata, file_name=None):
+        return real(metadata)
+
+    return unnamed
+
+
+def _only_candidate(test, item):
+    """The positive text of an anchored item's one candidate."""
+    evidence = json.loads(item["evidence"])
+    test.assertEqual(evidence.get("anchor"), "saver")
+    test.assertEqual(evidence.get("profile"), "inspection")
+    (candidate,) = evidence["candidates"]
+    return candidate["positive"]["text"]
+
+
+class TestInspectionItemsAnchor(InspectionTestCase):
+    async def test_file_its_saver_names_is_queued_with_the_anchor_choice(self):
+        self._roots()
+        path = self._write("branch_1_00001_.png", graph=branch_graph(*PROMPTS))
+
+        job = await self._scan()
+
+        (item,) = self._items()
+        self.assertEqual(_only_candidate(self, item), "add a red hat")
+        self.assertEqual(item["candidate_count"], 1)
+        self.assertEqual(
+            item["resolver_version"],
+            review_capture.inspection_review_version(
+                {"prompt": branch_graph(*PROMPTS)}
+            ),
+        )
+        (row,) = job["outcomes"]["review"]
+        self.assertEqual(row["path"], "branch_1_00001_.png")
+        # Queued, never linked: ingestion stays on the default profile.
+        self.assertIsNone(self._linked_text(path))
+
+    async def test_anchored_item_stays_on_rescan(self):
+        self._roots()
+        self._write("branch_1_00001_.png", graph=branch_graph(*PROMPTS))
+        await self._scan()
+        (item,) = self._items()
+
+        job = await self._scan("all")
+
+        self.assertEqual([i["id"] for i in self._items()], [item["id"]])
+        self.assertEqual(job["scope"]["review_untraced"], 0)
+
+    async def test_file_no_saver_names_stays_queued(self):
+        self._roots()
+        self._write("elsewhere_00001_.png", graph=branch_graph(*PROMPTS))
+
+        await self._scan()
+
+        (item,) = self._items()
+        self.assertEqual(item["candidate_count"], len(PROMPTS))
+        self.assertNotIn("anchor", json.loads(item["evidence"]))
+
+    async def test_file_several_savers_name_stays_queued(self):
+        self._roots()
+        graph = branch_graph(*PROMPTS, prefixes=["same", "same", "other"])
+        self._write("same_00001_.png", graph=graph)
+
+        await self._scan()
+
+        (item,) = self._items()
+        self.assertEqual(item["candidate_count"], len(PROMPTS))
+
+    async def test_old_rule_item_is_replaced_by_the_anchor_choice(self):
+        self._roots()
+        self._write("branch_1_00001_.png", graph=branch_graph(*PROMPTS))
+        real = review_capture.resolve_inspection
+        with (
+            patch.object(review_capture, "REVIEW_RULE_VERSION", 0),
+            patch.object(
+                review_capture, "resolve_inspection", side_effect=_without_name(real)
+            ),
+        ):
+            await self._scan()
+        (old,) = self._items()  # queued under the old rule with every branch
+        self.assertEqual(old["candidate_count"], len(PROMPTS))
+
+        await self._scan("all")
+
+        (item,) = self._items()
+        self.assertNotEqual(item["id"], old["id"])
+        self.assertEqual(_only_candidate(self, item), "add a red hat")
+
+    async def test_apply_links_the_anchor_choice(self):
+        self._require_routes()
+        self._roots()
+        path = self._write("branch_1_00001_.png", graph=branch_graph(*PROMPTS))
+        await self._scan()
+
+        _item, (status, data) = await self._apply("add a red hat")
+
+        self.assertEqual(status, 200, data)
+        self.assertEqual(self._linked_text(path), "add a red hat")
+
+    async def test_apply_refuses_an_old_choice_the_anchor_contradicts(self):
+        self._require_routes()
+        self._roots()
+        path = self._write("branch_1_00001_.png", graph=branch_graph(*PROMPTS))
+        real = review_capture.resolve_inspection
+        with (
+            patch.object(review_capture, "REVIEW_RULE_VERSION", 0),
+            patch.object(
+                review_capture, "resolve_inspection", side_effect=_without_name(real)
+            ),
+        ):
+            await self._scan()
+        item, detail, observed = await self._open()
+        # branch_1 is the saver of "add a red hat"; choose another branch.
+        wrong = self._candidate(detail, "make it night")
+
+        status, data = await self._decide(
+            item["id"], self._body(detail, observed, wrong["hash"])
+        )
+
+        self.assertEqual(status, 409, data)
+        self.assertEqual(data["code"], "review-conflict", data)
+        self.assertIsNone(self._linked_text(path))
+
+
+class TestDefaultItemsAnchor(InspectionTestCase):
+    async def test_anchor_resolves_and_nothing_links_so_it_is_queued(self):
+        """Maintainer ruling: never unlinked and unqueued. The anchor resolves
+        the file, the default profile and the parser floor link nothing, so
+        it is queued with the anchor's pair as its choice."""
+        self._roots()
+        path = self.output_dir / "side_0_00001_.png"
+        graph = saved_unlinkable_graph("cat", "dog")
+        write_png(path, prompt=graph)
+        self.assertEqual(
+            review_capture.resolve_prompt_pair({"prompt": graph})["state"], "ambiguous"
+        )
+
+        job = await self._scan()
+
+        (item,) = self._items()
+        self.assertEqual(_only_candidate(self, item), "cat")
+        (row,) = job["outcomes"]["review"]
+        self.assertIn("more than one prompt could match", row["detail"])
+        self.assertIsNone(self._linked_text(path))
+
+    async def test_default_ambiguous_file_no_saver_names_stays_queued(self):
+        self._roots()
+        write_png(
+            self.output_dir / "unsaved_00001_.png",
+            prompt=saved_unlinkable_graph("cat", "dog"),
+        )
+
+        await self._scan()
+
+        (item,) = self._items()
+        evidence = json.loads(item["evidence"])
+        self.assertNotIn("profile", evidence)
+        self.assertNotIn("anchor", evidence)
+        self.assertEqual(item["resolver_version"], review_capture.RESOLVER_VERSION)
+
+    async def test_file_the_parser_floor_links_is_not_queued(self):
+        """Already linked by the default path: unchanged by the anchor."""
+        self._roots()
+        graph = saved_unlinkable_graph("cat", "dog")
+        del graph["3"]  # the parser floor now reads a real sampler
+        path = self.output_dir / "side_0_00001_.png"
+        write_png(path, prompt=graph)
+
+        await self._scan()
+
+        self.assertEqual(self._items(), [])
+        self.assertIsNotNone(self._linked_text(path))
+
+    async def test_apply_refuses_an_old_default_choice_the_anchor_contradicts(self):
+        self._require_routes()
+        self._roots()
+        path = self.output_dir / "side_0_00001_.png"
+        write_png(path, prompt=saved_unlinkable_graph("cat", "dog"))
+        with patch.object(review_capture, "_captured_inspection", return_value=None):
+            await self._scan()  # queued from the default profile, as before #390
+        item, detail, observed = await self._open()
+        # side_0 is the saver of "cat"; choose the other branch.
+        wrong = self._candidate(detail, "dog")
+
+        status, data = await self._decide(
+            item["id"], self._body(detail, observed, wrong["hash"])
+        )
+
+        self.assertEqual(status, 409, data)
+        self.assertEqual(data["code"], "review-conflict", data)
+        self.assertIsNone(self._linked_text(path))
+
+    async def test_old_default_item_is_replaced_by_the_anchor_choice(self):
+        self._roots()
+        write_png(
+            self.output_dir / "side_0_00001_.png",
+            prompt=saved_unlinkable_graph("cat", "dog"),
+        )
+        with patch.object(review_capture, "_captured_inspection", return_value=None):
+            await self._scan()  # queued from the default profile, as before #390
+        (old,) = self._items()
+        self.assertEqual(old["resolver_version"], review_capture.RESOLVER_VERSION)
+
+        await self._scan("all")
+
+        (item,) = self._items()
+        self.assertNotEqual(item["id"], old["id"])
+        self.assertEqual(_only_candidate(self, item), "cat")
+
+        with _counted(review_capture, "resolve_inspection") as calls:
+            await self._scan("all")  # now queued under its version: no re-read
+        self.assertEqual(calls, [])
+        self.assertEqual([i["id"] for i in self._items()], [item["id"]])
+
+    async def test_decided_anchored_item_is_kept(self):
+        self._require_routes()
+        self._roots()
+        write_png(
+            self.output_dir / "side_0_00001_.png",
+            prompt=saved_unlinkable_graph("cat", "dog"),
+        )
+        await self._scan()
+        item, detail, observed = await self._open()
+        status, data = await self._decide(
+            item["id"], self._body(detail, observed, "unresolved")
+        )
+        self.assertEqual(status, 200, data)
+        before = (self._items(), self._decisions())
+
+        await self._scan("all")
+
+        self.assertEqual((self._items(), self._decisions()), before)
+
+    async def test_apply_accepts_an_old_default_choice_the_anchor_names(self):
+        self._require_routes()
+        self._roots()
+        path = self.output_dir / "side_0_00001_.png"
+        write_png(path, prompt=saved_unlinkable_graph("cat", "dog"))
+        with patch.object(review_capture, "_captured_inspection", return_value=None):
+            await self._scan()
+
+        _item, (status, data) = await self._apply("cat")
+
+        self.assertEqual(status, 200, data)
+        self.assertEqual(self._linked_text(path), "cat")
+
+
+class TestSaverAnchored(InspectionTestCase):
+    def test_only_the_anchor_counts(self):
+        anchored = {"prompt": saved_unlinkable_graph("cat", "dog")}
+        self.assertTrue(review_capture.saver_anchored(anchored, "side_1_00001_.png"))
+        self.assertFalse(review_capture.saver_anchored(anchored, "nope_00001_.png"))
+        self.assertFalse(review_capture.saver_anchored(anchored, None))
+        # One sampler resolves without any name: the anchor decides nothing.
+        single = {"prompt": branch_graph("make it night")}
+        self.assertFalse(review_capture.saver_anchored(single, "branch_0_00001_.png"))

tests/test_utils_logging_config.py

@@ -709,7 +709,8 @@ class TestRotatedCopyRule(IsolatedLoggerTestCase):
         self.assertNotEqual(link["size"], os.path.getsize(target))
         for row in rows.values():
             self.assertEqual(
-                set(row), {"filename", "path", "size", "modified", "is_main", "kind"}
+                set(row),
+                {"filename", "path", "size", "modified", "is_main", "kind", "identity"},
             )
 
     def test_an_unreadable_timestamp_does_not_break_the_scan(self):
@@ -763,6 +764,217 @@ class TestRotatedCopyRule(IsolatedLoggerTestCase):
                     self.manager.log_file_entry(name)
 
 
+class TestClearOnlyWhatWasConfirmed(IsolatedLoggerTestCase):
+    """#389: Clear all logs empties or deletes no more than its confirmation showed."""
+
+    def setUp(self):
+        super().setUp()
+        self.logger.info("active log line")  # the active log exists
+        self.outside = os.path.join(self.tmp.name, "outside")
+        os.makedirs(self.outside)
+
+    def _path(self, name):
+        return os.path.join(self.log_dir, name)
+
+    def _write(self, name, data=b"rotated\n"):
+        with open(self._path(name), "wb") as handle:
+            handle.write(data)
+
+    def _read(self, path):
+        with open(path, "rb") as handle:
+            return handle.read()
+
+    def _snapshot(self):
+        return {name: self._read(self._path(name)) for name in os.listdir(self.log_dir)}
+
+    def _assert_unchanged(self, before, growing="prompt_manager.log"):
+        """Nothing deleted or emptied (the refusal itself logs to ``growing``)."""
+        after = self._snapshot()
+        self.assertEqual(sorted(after), sorted(before))
+        for name, data in before.items():
+            if name == growing:
+                self.assertTrue(after[name].startswith(data), name)
+            else:
+                self.assertEqual(after[name], data, name)
+
+    def _confirmed(self):
+        """What the workspace dialog showed: the listed active and rotated rows."""
+        return [
+            {"name": row["filename"], "identity": row["identity"]}
+            for row in self.manager.get_log_files()
+            if row["kind"] in ("active", "rotated")
+        ]
+
+    def _file_handler(self):
+        return next(
+            h
+            for h in self.manager.logger.handlers
+            if isinstance(h, logging.FileHandler)
+        )
+
+    def test_listed_identity_is_a_string_that_survives_json_numbers(self):
+        # mtime_ns (~1.7e18) and many inodes exceed 2**53: as JSON numbers a
+        # browser would round them and every confirmed clear would be refused.
+        self._write("prompt_manager.log.1")
+        for row in self.manager.get_log_files():
+            with self.subTest(name=row["filename"]):
+                self.assertIsInstance(row["identity"], str)
+                self.assertRegex(row["identity"], r"^\d+:-?\d+:\d+$")
+
+    def test_a_copy_dated_before_1970_can_be_confirmed(self):
+        self._write("prompt_manager.log.1")
+        try:
+            os.utime(self._path("prompt_manager.log.1"), ns=(0, -(10**9)))
+        except (OSError, OverflowError, ValueError) as error:
+            self.skipTest(f"this filesystem cannot store a time before 1970: {error}")
+        confirmed = self._confirmed()
+        if not any(":-" in row["identity"] for row in confirmed):
+            self.skipTest("the filesystem did not keep the negative time")
+        result = self.manager.truncate_logs(confirmed)
+        self.assertIn("prompt_manager.log.1", result["truncated"])
+
+    def test_a_rollover_after_the_confirmation_deletes_nothing(self):
+        self._write("prompt_manager.log.1", b"one" * 10)
+        self._write("prompt_manager.log.2", b"two" * 10)
+        confirmed = self._confirmed()
+        self.logger.info("written after the confirmation")
+        self._file_handler().doRollover()  # .2 -> .3, .1 -> .2, log -> .1
+        before = self._snapshot()
+        with self.assertRaises(logging_config.ClearRefused) as caught:
+            self.manager.truncate_logs(confirmed)
+        self.assertEqual(caught.exception.code, "logs-changed")
+        self._assert_unchanged(before)
+
+    def test_a_copy_that_appeared_after_the_confirmation_is_kept(self):
+        self._write("prompt_manager.log.1", b"one")
+        confirmed = self._confirmed()
+        self._write("prompt_manager.log.5", b"not shown")
+        result = self.manager.truncate_logs(confirmed)
+        self.assertEqual(result["errors"], [])
+        self.assertEqual(
+            sorted(result["truncated"]), ["prompt_manager.log", "prompt_manager.log.1"]
+        )
+        self.assertEqual(
+            sorted(os.listdir(self.log_dir)),
+            ["prompt_manager.log", "prompt_manager.log.5"],
+        )
+
+    def _clear(self, confirmed=None):
+        """Run a clear and return what it raised (None), so the files are
+        asserted first: on the old code the test fails on lost bytes."""
+        try:
+            if confirmed is None:
+                self.manager.truncate_logs()  # the legacy page's call
+            else:
+                self.manager.truncate_logs(confirmed)
+        except Exception as error:  # noqa: BLE001 - asserted below
+            return error
+        return None
+
+    def test_a_foreign_append_to_the_active_log_is_refused(self):
+        # A second ComfyUI sharing the folder writes the same active log.
+        self._write("prompt_manager.log.1")
+        listings = (None, "listed")
+        for listing in listings:
+            with self.subTest(listed=listing is not None):
+                confirmed = self._confirmed() if listing else None
+                with open(self._path("prompt_manager.log"), "ab") as other:
+                    other.write(b"line from another process\n")
+                before = self._snapshot()
+                error = self._clear(confirmed)
+                self._assert_unchanged(before)
+                self.assertEqual(getattr(error, "code", None), "logs-foreign-writer")
+                self.logger.info("our own line")  # our stream catches up again
+
+    @unittest.skipIf(os.name == "nt", "Windows cannot rename a log that is open")
+    def test_a_foreign_rotation_of_our_log_is_refused(self):
+        # Another process rolled the shared log over: our handler still writes
+        # the file now named .1, and a new active log is the other's.
+        os.rename(self._path("prompt_manager.log"), self._path("prompt_manager.log.1"))
+        self._write("prompt_manager.log", b"the other process's log\n")
+        before = self._snapshot()
+        with self.assertRaises(logging_config.ClearRefused) as caught:
+            self.manager.truncate_logs()
+        self.assertEqual(caught.exception.code, "logs-foreign-writer")
+        # Our handler still writes the renamed file, so the refusal lands there.
+        self._assert_unchanged(before, growing="prompt_manager.log.1")
+
+    def test_a_copy_that_changes_mid_clear_stops_the_clear(self):
+        # The identity is checked again just before each unlink: another
+        # process rotating while we delete stops the clear there.
+        for n in (1, 2):
+            self._write(f"prompt_manager.log.{n}", b"r" * n)
+        confirmed = self._confirmed()
+        real_unlink = os.unlink
+        calls = []
+
+        def unlink_then_rotate(path, *args, **kwargs):
+            real_unlink(path, *args, **kwargs)
+            if not calls:
+                with open(self._path("prompt_manager.log.2"), "ab") as handle:
+                    handle.write(b"rotated in by another process")
+            calls.append(os.path.basename(path))
+
+        active_before = self._read(self._path("prompt_manager.log"))
+        with mock.patch.object(logging_config.os, "unlink", unlink_then_rotate):
+            result = self.manager.truncate_logs(confirmed)
+        self.assertEqual(calls, ["prompt_manager.log.1"])
+        self.assertEqual(result["truncated"], ["prompt_manager.log.1"])
+        self.assertEqual(len(result["errors"]), 1)
+        self.assertIn("prompt_manager.log.2", result["errors"][0])
+        self.assertTrue(os.path.exists(self._path("prompt_manager.log.2")))
+        # The active log is emptied last, so it is kept too.
+        self.assertTrue(
+            self._read(self._path("prompt_manager.log")).startswith(active_before)
+        )
+
+    def _require_hard_link(self, source, link):
+        try:
+            os.link(source, link)
+        except (OSError, NotImplementedError, AttributeError) as error:
+            self.skipTest(f"this filesystem cannot make hard links: {error}")
+
+    def test_a_hard_linked_active_log_is_never_emptied(self):
+        other_name = os.path.join(self.outside, "kept-by-the-user.log")
+        self._require_hard_link(self._path("prompt_manager.log"), other_name)
+        kept = self._read(other_name)
+        self.assertTrue(kept)
+        for listed in (False, True):
+            with self.subTest(listed=listed):
+                error = self._clear(self._confirmed() if listed else None)
+                self.assertTrue(self._read(other_name).startswith(kept))
+                self.assertEqual(getattr(error, "code", None), "logs-hard-link")
+
+    def test_a_hard_linked_active_log_with_file_logging_off_is_kept(self):
+        self.manager.update_config({"file_logging": False})
+        other_name = os.path.join(self.outside, "kept-by-the-user.log")
+        self._require_hard_link(self._path("prompt_manager.log"), other_name)
+        kept = self._read(other_name)
+        error = self._clear()
+        self.assertEqual(self._read(other_name), kept)
+        self.assertEqual(getattr(error, "code", None), "logs-hard-link")
+
+    def test_a_malformed_confirmation_is_refused_before_anything_changes(self):
+        self._write("prompt_manager.log.1")
+        before = self._snapshot()
+        for expected in (
+            "prompt_manager.log.1",
+            [{"name": "prompt_manager.log.1"}],
+            [{"name": "prompt_manager.log.1", "identity": 7}],
+            [{"name": "prompt_manager.log.1", "identity": "1:2"}],
+            [{"name": 1, "identity": "1:2:3"}],
+            [
+                {"name": "prompt_manager.log.1", "identity": "1:2:3"},
+                {"name": "prompt_manager.log.1", "identity": "1:2:3"},
+            ],
+            ["prompt_manager.log.1"],
+        ):
+            with self.subTest(expected=expected):
+                with self.assertRaises(ValueError):
+                    self.manager.truncate_logs(expected)
+                self._assert_unchanged(before)
+
+
 class TestConfiguration(IsolatedLoggerTestCase):
     def test_in_flight_record_cannot_reopen_retired_file_handler(self) -> None:
         handler = next(

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

utils/exit_signal.py

@@ -0,0 +1,110 @@
+"""The process exit signal: running work stops at its next safe point (#376).
+
+ComfyUI catches Ctrl+C in ``main.py`` and returns without closing its event
+loop. The interpreter then runs ``threading._shutdown``, which first calls the
+``threading._register_atexit`` callbacks, newest first. One of them,
+``concurrent.futures``'s ``_python_exit``, joins every executor thread, and
+only after that do the ``atexit`` callbacks run (the package teardown). A call
+still running on the default executor (a scan batch, a model download, a
+thumbnail batch) used to keep the process alive until it returned.
+
+``install()`` registers a callback in that same list after ``concurrent.futures``
+did, so it runs before the join. It sets the signal that ``requested()`` reads
+(``Job.cancelled`` and ``map_parallel`` read it) and runs ``on_exit``.
+
+``threading._register_atexit`` is CPython-internal. It has the same contract
+on 3.12, 3.13 and 3.14. Without it, or when it refuses the registration, exit
+behaves as before: the call in flight runs to its end.
+"""
+
+import logging
+import threading
+
+logger = logging.getLogger("prompt_manager.exit")
+
+_requested = threading.Event()
+_install_lock = threading.Lock()
+_installed = False
+_on_exit = []
+
+
+class ExitRequested(Exception):
+    """The interpreter is exiting: work not yet started is not started."""
+
+
+def requested():
+    """True once the interpreter has started exiting."""
+    return _requested.is_set()
+
+
+def request():
+    """Set the signal (the exit hook; tests)."""
+    _requested.set()
+
+
+def checkpoint():
+    """Raise ExitRequested once the interpreter is exiting."""
+    if _requested.is_set():
+        raise ExitRequested("ComfyUI is exiting")
+
+
+def checked(func):
+    """``func`` behind a checkpoint: a call that starts after the signal
+    raises ExitRequested instead of running."""
+
+    def run(*args, **kwargs):
+        checkpoint()
+        return func(*args, **kwargs)
+
+    return run
+
+
+def _hook():
+    """The threading atexit callback. It never raises: an exception here
+    would end ``threading._shutdown`` before it shuts the executors."""
+    _requested.set()
+    for action in tuple(_on_exit):
+        try:
+            action()
+        except (Exception, KeyboardInterrupt):
+            logger.error("Stopping work at exit failed", exc_info=True)
+
+
+def install(on_exit=None):
+    """Register the exit hook once per process; True when it is registered.
+
+    ``on_exit`` (optional) runs inside the hook, after the signal is set.
+    False (logged at DEBUG) when this Python has no ``_register_atexit`` or
+    refuses it; exit then waits for running work as before.
+    """
+    global _installed
+    with _install_lock:
+        if on_exit is not None and on_exit not in _on_exit:
+            _on_exit.append(on_exit)
+        if _installed:
+            return True
+        # Imported first so its pool join is registered before our hook:
+        # the callbacks run newest first, so ours runs before that join.
+        import concurrent.futures.thread  # noqa: F401
+
+        register = getattr(threading, "_register_atexit", None)
+        if not callable(register):
+            logger.debug("threading._register_atexit is missing; exit waits for jobs")
+            return False
+        try:
+            register(_hook)
+        except RuntimeError as error:  # already shutting down
+            logger.debug(f"Exit hook not registered: {error}")
+            return False
+        _installed = True
+        return True
+
+
+def reset():
+    """Clear the signal and forget the registration (tests only; a hook
+    already registered stays in ``threading`` and sets the signal at exit)."""
+    global _installed
+    with _install_lock:
+        _requested.clear()
+        _installed = False
+        _on_exit.clear()

utils/logging_config.py

@@ -39,6 +39,35 @@ class RotatedChanged(Exception):
         self.name = name
 
 
+# Clear all logs (#389): why it changed nothing, as the API and the pages say it.
+CLEAR_CHANGED = (
+    "The log files changed since you confirmed; nothing was cleared."
+    " Read the list again and clear again."
+)
+CLEAR_FOREIGN_WRITER = (
+    "Another program is writing to prompt_manager.log, such as a second ComfyUI"
+    " sharing this log folder; nothing was cleared. Stop it and clear again."
+)
+CLEAR_HARD_LINK = (
+    "prompt_manager.log has another name (a hard link); emptying it would empty"
+    " that file too, so nothing was cleared. Remove the other name, then clear"
+    " again."
+)
+# A confirmed row's identity: "<size>:<mtime_ns>:<ino>", a string because
+# mtime_ns and many inodes are larger than a browser's exact integers.
+# mtime_ns is negative for a time before 1970 (a restored archive, say).
+_IDENTITY = re.compile(r"([0-9]{1,20}):(-?[0-9]{1,20}):([0-9]{1,40})")
+MAX_CONFIRMED_ROWS = 10000
+
+
+class ClearRefused(Exception):
+    """Clear all logs changed nothing; ``code`` names why, the message says it."""
+
+    def __init__(self, code: str, message: str):
+        super().__init__(message)
+        self.code = code
+
+
 def _name_key(name: str) -> str:
     """How a name compares: case-folded where the filesystem folds it (Windows)."""
     return os.path.normcase(name)
@@ -87,35 +116,37 @@ def scan_log_folder(log_dir) -> List[Dict[str, Any]]:
     be read is left out. Never logs: the cleanup calls it while it holds the
     file handler's lock.
     """
-    rows = []
     try:
         names = [entry.name for entry in os.scandir(log_dir)]
     except OSError:
-        return rows
-    for name in names:
-        path = os.path.join(log_dir, name)
-        try:
-            st = os.lstat(path)
-            kind = _kind(name, path, st)
-        except OSError:
-            continue
-        try:
-            modified = datetime.fromtimestamp(st.st_mtime).isoformat()
-        except (OSError, OverflowError, ValueError):  # a time before 1970 on Windows
-            modified = ""
-        rows.append(
-            {
-                "name": name,
-                "kind": kind,
-                "size": st.st_size,
-                "mtime_ns": st.st_mtime_ns,
-                "ino": st.st_ino,
-                "modified": modified,
-                "_dir": stat.S_ISDIR(st.st_mode) and kind != "link",
-                "_regular": stat.S_ISREG(st.st_mode) and kind != "link",
-            }
-        )
-    return rows
+        return []
+    rows = (scan_log_entry(log_dir, name) for name in names)
+    return [row for row in rows if row is not None]
+
+
+def scan_log_entry(log_dir, name) -> Optional[Dict[str, Any]]:
+    """One entry's scan row (``lstat``), or None when it cannot be read."""
+    path = os.path.join(log_dir, name)
+    try:
+        st = os.lstat(path)
+        kind = _kind(name, path, st)
+    except OSError:
+        return None
+    try:
+        modified = datetime.fromtimestamp(st.st_mtime).isoformat()
+    except (OSError, OverflowError, ValueError):  # a time before 1970 on Windows
+        modified = ""
+    return {
+        "name": name,
+        "kind": kind,
+        "size": st.st_size,
+        "mtime_ns": st.st_mtime_ns,
+        "ino": st.st_ino,
+        "modified": modified,
+        "_dir": stat.S_ISDIR(st.st_mode) and kind != "link",
+        "_regular": stat.S_ISREG(st.st_mode) and kind != "link",
+        "_nlink": st.st_nlink,
+    }
 
 
 def public_row(row: Dict[str, Any]) -> Dict[str, Any]:
@@ -127,6 +158,37 @@ def rotated_sort_key(row: Dict[str, Any]) -> int:
     return _rotated_number(row["name"]) or 0
 
 
+def file_identity(row: Dict[str, Any]) -> str:
+    """A listed file's identity as the client sends it back: size, mtime_ns, ino."""
+    return f"{row['size']}:{row['mtime_ns']}:{row['ino']}"
+
+
+def parse_confirmed(expected) -> List[Dict[str, Any]]:
+    """The rows a Clear all logs confirmation showed, as review rows.
+
+    ``expected`` is a list of ``{name, identity}`` objects from
+    ``/logs/files``; anything else raises ValueError.
+    """
+    if not isinstance(expected, list) or len(expected) > MAX_CONFIRMED_ROWS:
+        raise ValueError("expected must be a list of listed log files")
+    rows, seen = [], set()
+    for item in expected:
+        if not isinstance(item, dict):
+            raise ValueError("each expected file must be an object")
+        name, identity = item.get("name"), item.get("identity")
+        if not isinstance(name, str) or not isinstance(identity, str):
+            raise ValueError("each expected file needs a name and an identity")
+        match = _IDENTITY.fullmatch(identity)
+        if match is None:
+            raise ValueError("an expected file's identity is not size:mtime_ns:ino")
+        size, mtime_ns, ino = (int(part) for part in match.groups())
+        if _name_key(name) in seen:
+            raise ValueError("an expected file is listed twice")
+        seen.add(_name_key(name))
+        rows.append({"name": name, "size": size, "mtime_ns": mtime_ns, "ino": ino})
+    return rows
+
+
 def log_file_entry(log_dir, name) -> Path:
     """The path of a downloadable log file named ``name`` (blocking).
 
@@ -157,6 +219,11 @@ def _stream_is(stream, ino: int) -> bool:
     return not (ino and open_ino and open_ino != ino)
 
 
+def _stream_ino_is(opened: os.stat_result, ino: int) -> bool:
+    """Whether a stream's ``fstat`` is the file with inode ``ino`` (0: unknown)."""
+    return not (ino and opened.st_ino and opened.st_ino != ino)
+
+
 def _error_text(error: BaseException) -> str:
     """An OSError's reason and file name, without the folder it is in."""
     if isinstance(error, OSError):
@@ -432,6 +499,7 @@ class PromptManagerLogger:
                     "modified": row["modified"],
                     "is_main": row["kind"] == "active",
                     "kind": row["kind"],
+                    "identity": file_identity(row),
                 }
             )
 
@@ -530,8 +598,14 @@ class PromptManagerLogger:
             finally:
                 handler.release()
         if not handled:
-            flags = os.O_WRONLY | os.O_TRUNC | getattr(os, "O_NOFOLLOW", 0)
-            os.close(os.open(str(log_file), flags))
+            flags = os.O_WRONLY | getattr(os, "O_NOFOLLOW", 0)
+            fd = os.open(str(log_file), flags)
+            try:
+                if os.fstat(fd).st_nlink > 1:  # emptying it would empty the other
+                    raise ClearRefused("logs-hard-link", CLEAR_HARD_LINK)
+                os.ftruncate(fd, 0)
+            finally:
+                os.close(fd)
 
     def delete_rotated(self, expected: List[Dict[str, Any]]) -> Dict[str, Any]:
         """Delete exactly the reviewed rotated copies, or none of them (blocking).
@@ -582,47 +656,126 @@ class PromptManagerLogger:
         )
         return {"removed": removed, "failed": failed}
 
-    def truncate_logs(self) -> Dict[str, Any]:
-        """Empty the active log and delete its rotated copies (Rule R).
+    def _active_streams(self, active: Optional[Dict[str, Any]]):
+        """``fstat`` and offset of every handler stream writing the active name."""
+        target = os.path.normcase(
+            os.path.realpath(os.path.join(self.log_dir, ACTIVE_LOG_NAME))
+        )
+        for handler in self._file_handlers():
+            stream = getattr(handler, "stream", None)
+            base = getattr(handler, "baseFilename", None)
+            if stream is None or base is None:
+                continue
+            if os.path.normcase(os.path.realpath(base)) != target:
+                continue
+            try:
+                stream.flush()
+                fd = stream.fileno()
+                yield os.fstat(fd), os.lseek(fd, 0, os.SEEK_CUR)
+            except (OSError, ValueError, AttributeError):
+                continue
 
-        Every other file in the log folder, and every link, is kept. Also
-        clears the memory buffer. Runs under ``_files_lock`` and the file
-        handlers' locks, so no rollover recreates a just-deleted copy, and
-        logs only once they are released.
+    def _check_clear_is_safe(self, rows, active) -> None:
+        """Refuse a clear that would empty or delete what nobody confirmed.
+
+        Called under the files lock and every handler lock, so nothing of this
+        process writes or rolls over meanwhile; what changes is another
+        program's doing. Our handler's stream is flushed: a file larger than
+        the stream's own offset was appended to by someone else, and a stream
+        whose file is now a rotated copy was rolled over by someone else.
+        Either is a second writer (another ComfyUI sharing the folder). This
+        is best effort, not a lock: writes before our own last write, or any
+        write while this process has file logging off, go unseen. An
+        active log with more than one name (a hard link) is never emptied: the
+        other name would be emptied too.
+        """
+        rotated = {row["ino"] for row in rows if row["kind"] == "rotated"}
+        rotated.discard(0)
+        nlink = active["_nlink"] if active is not None else 1
+        for opened, offset in self._active_streams(active):
+            if (opened.st_ino and opened.st_ino in rotated) or opened.st_size > offset:
+                raise ClearRefused("logs-foreign-writer", CLEAR_FOREIGN_WRITER)
+            if active is not None and _stream_ino_is(opened, active["ino"]):
+                nlink = max(nlink, opened.st_nlink)
+        if nlink > 1:
+            raise ClearRefused("logs-hard-link", CLEAR_HARD_LINK)
+
+    @staticmethod
+    def _confirmed_targets(confirmed, current, active) -> List[Dict[str, Any]]:
+        """The rotated copies a confirmation listed, each as it was listed.
+
+        Refuses (``logs-changed``) when any listed copy is missing or differs
+        in kind, size, ``mtime_ns`` or inode, or when the listed active log is
+        now another file (it was rolled over). A copy nobody listed is kept.
+        """
+        targets = []
+        for reviewed in confirmed:
+            key = _name_key(reviewed["name"])
+            row = current.get(key)
+            if key == _name_key(ACTIVE_LOG_NAME):
+                ino = reviewed["ino"]
+                if active is None or (ino and active["ino"] and active["ino"] != ino):
+                    raise ClearRefused("logs-changed", CLEAR_CHANGED)
+                continue
+            if _rotated_number(reviewed["name"]) is None:
+                continue  # never deleted, listed or not
+            if not _matches_review(row, reviewed):
+                raise ClearRefused("logs-changed", CLEAR_CHANGED)
+            targets.append(row)
+        return targets
+
+    def truncate_logs(self, expected=None) -> Dict[str, Any]:
+        """Empty the active log and delete its rotated copies (Rule R, #389).
+
+        ``expected`` is what the confirmation showed: ``{name, identity}``
+        rows of ``/logs/files`` (``parse_confirmed``; ValueError when
+        malformed). Then only the listed rotated copies are deleted, and only
+        when each still has the listed size, ``mtime_ns`` and inode and the
+        active log is still the listed file; otherwise ClearRefused
+        (``logs-changed``) with nothing changed. Without it (the legacy admin
+        page, whose confirmation lists nothing) every rotated copy present now
+        is deleted.
+
+        Either way, a second writer of the active log (``logs-foreign-writer``)
+        or an active log with a hard link (``logs-hard-link``) refuses the
+        whole clear first. Each copy's identity is checked again just before
+        its unlink; one that changed (another process rolling over meanwhile)
+        stops the clear there, keeping it, the rest and the active log, which
+        is emptied last. Every other file in the log folder, and every link,
+        is kept. Also clears the memory buffer. Runs under ``_files_lock`` and
+        the file handlers' locks and logs only once they are released.
 
         Returns:
             Dictionary with:
             - truncated: List of successfully truncated files
             - errors: List of error messages for failed operations
         """
+        confirmed = None if expected is None else parse_confirmed(expected)
         results = {"truncated": [], "errors": []}
+        refused = None
 
         with self._files_lock:
             held = self._acquire_file_handlers()
             try:
                 rows = scan_log_folder(self.log_dir)
-                active = [row for row in rows if row["kind"] == "active"]
-                rotated = sorted(
-                    (row for row in rows if row["kind"] == "rotated"),
-                    key=rotated_sort_key,
-                )
-                for row in active + rotated:
-                    path = os.path.join(self.log_dir, row["name"])
-                    try:
-                        if row["kind"] == "active":
-                            self._truncate_main_log(Path(path), row["ino"])
-                        else:
-                            os.unlink(path)
-                        results["truncated"].append(row["name"])
-                    except OSError as error:
-                        results["errors"].append(
-                            f"Failed to truncate {row['name']}: {_error_text(error)}"
-                        )
-                with self._buffer_lock:
-                    self._log_buffer.clear()
+                current = {_name_key(row["name"]): row for row in rows}
+                active = next((row for row in rows if row["kind"] == "active"), None)
+                try:
+                    if confirmed is None:
+                        targets = [row for row in rows if row["kind"] == "rotated"]
+                    else:
+                        targets = self._confirmed_targets(confirmed, current, active)
+                    self._check_clear_is_safe(rows, active)
+                except ClearRefused as error:
+                    refused = error
+                else:
+                    self._clear(sorted(targets, key=rotated_sort_key), active, results)
             finally:
                 self._release_handlers(held)
 
+        if refused is not None:
+            self.logger.warning("Clear all logs refused (%s)", refused.code)
+            raise refused
         if results["truncated"]:
             self.logger.info("Truncated log files: %s", ", ".join(results["truncated"]))
         for error_msg in results["errors"]:
@@ -630,6 +783,39 @@ class PromptManagerLogger:
 
         return results
 
+    def _clear(self, targets, active, results) -> None:
+        """Delete ``targets`` (each re-checked) then empty ``active`` (locks held)."""
+        for reviewed in targets:
+            name = reviewed["name"]
+            if not _matches_review(scan_log_entry(self.log_dir, name), reviewed):
+                results["errors"].append(
+                    f"{name} changed while clearing (another program may be writing"
+                    " to this log folder); it, the copies after it and the active"
+                    " log were kept"
+                )
+                return
+            try:
+                os.unlink(os.path.join(self.log_dir, name))
+                results["truncated"].append(name)
+            except OSError as error:
+                results["errors"].append(
+                    f"Failed to truncate {name}: {_error_text(error)}"
+                )
+        if active is not None:
+            try:
+                self._truncate_main_log(
+                    Path(self.log_dir, active["name"]), active["ino"]
+                )
+                results["truncated"].append(active["name"])
+            except ClearRefused as error:
+                results["errors"].append(str(error))
+            except OSError as error:
+                results["errors"].append(
+                    f"Failed to truncate {active['name']}: {_error_text(error)}"
+                )
+        with self._buffer_lock:
+            self._log_buffer.clear()
+
     def update_config(self, new_config: Dict[str, Any]):
         """Update logging configuration.
 

utils/parallel.py

@@ -21,6 +21,8 @@ import logging
 import threading
 from concurrent.futures import ThreadPoolExecutor
 
+from .exit_signal import checked
+
 SHUTDOWN_REFUSAL = "interpreter shutdown"
 _logger = logging.getLogger("prompt_manager.parallel")
 _noted = set()
@@ -49,8 +51,11 @@ def map_parallel(func, items, workers):
     ``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. A pool refused at interpreter
-    exit runs the items it did not start on the calling thread (#375).
+    exit runs the items it did not start on the calling thread (#375). An item
+    that has not started when the interpreter starts exiting raises
+    ExitRequested instead (#376).
     """
+    func = checked(func)
     items = list(items)
     workers = max(1, int(workers or 1))
     if workers == 1 or len(items) <= 1:

utils/review_capture.py

@@ -40,6 +40,19 @@ while it is pending and undecided (never beside a decision, see
 profile the item recorded (``captured_review(..., item_profile(evidence))``).
 The inspection profile only ever produces review choices; ingestion's own
 links stay on the default profile, which never loads the table (D1).
+
+The saver anchor (#390): the inspection profile resolves a file with its
+name, as the workspace inspector does since #383, so the one saver whose
+``filename_prefix`` names the file anchors its branch. When that anchor alone
+resolves a file the default profile left ambiguous and nothing links
+(``saver_anchored``), the scan queues it with the anchor's pair as its one
+candidate (evidence ``anchor: "saver"``), never linking it: a file never
+falls through both the link and the queue (maintainer ruling). No match,
+several matches or an unfinished walk anchor nothing and the file is queued
+as before. Apply re-resolves with the file's name, so only the anchored
+branch can be applied. ``REVIEW_RULE_VERSION`` is folded into inspection
+item versions, so inspection items recorded before the anchor are recorded
+again by the next scan.
 """
 
 from __future__ import annotations
@@ -104,10 +117,16 @@ MAX_CANDIDATE_DEPTH = 8
 # The resolver profile an item's evidence was built with (``profile``).
 DEFAULT = "default"
 INSPECTION = "inspection"
+# The ``anchor`` an item recorded from the saver anchor names (#390).
+SAVER_ANCHOR = "saver"
+ANCHOR_REASON = "The saver that wrote this file names its branch"
 # Inspection item versions: above every plausible ``RESOLVER_VERSION`` and,
 # with 40 bits of the folded digest, below 2**53 (exact as a JSON number).
 INSPECTION_VERSION_BASE = 1 << 40
 _INSPECTION_VERSION_HEX = 10
+# How the scan and Apply use the inspection profile, folded into inspection
+# item versions. 1: they resolve with the file's name (the saver anchor, #390).
+REVIEW_RULE_VERSION = 1
 
 
 # Above this many distinct node classes the whole table is the fingerprint.
@@ -154,38 +173,92 @@ def inspection_review_version(metadata: Any = None) -> int | None:
     """The ``resolver_version`` of an item recorded from the inspection profile.
 
     Folds the profile with its budgets, ``RESOLVER_VERSION``,
-    ``INSPECTION_RESOLVER_VERSION`` and ``semantics_version(metadata)``, so
-    a change of either resolver, of the profile's bounds or of the table
-    about the graph's classes moves it. None when the table cannot be loaded.
+    ``INSPECTION_RESOLVER_VERSION``, ``REVIEW_RULE_VERSION`` and
+    ``semantics_version(metadata)``, so a change of either resolver, of how
+    review uses the profile, of the profile's bounds or of the table about
+    the graph's classes moves it. None when the table cannot be loaded.
     """
     table = semantics_version(metadata)
     if table is None:
         return None
     profile = repr(astuple(INSPECTION_PROFILE))
     seed = f"{profile}:{RESOLVER_VERSION}:{INSPECTION_RESOLVER_VERSION}:{table}"
+    if REVIEW_RULE_VERSION:  # 0 is the seed items had before the rule existed
+        seed += f":review-rule-{REVIEW_RULE_VERSION}"
     folded = hashlib.sha256(seed.encode("utf-8")).hexdigest()
     return INSPECTION_VERSION_BASE + int(folded[:_INSPECTION_VERSION_HEX], 16)
 
 
-def resolve_inspection(metadata: dict) -> dict:
-    """``metadata`` resolved with the inspection profile (bounded work)."""
-    return resolve_prompt_pair(metadata, profile=INSPECTION_PROFILE)
+def resolve_inspection(metadata: dict, file_name: str | None = None) -> dict:
+    """``metadata`` resolved with the inspection profile (bounded work).
+
+    ``file_name`` (the file's last path part) anchors the file on the one
+    saver that wrote it, as in the workspace inspector (#390).
+    """
+    return resolve_prompt_pair(
+        metadata, profile=INSPECTION_PROFILE, file_name=file_name
+    )
 
 
-def inspection_choice(metadata: dict) -> dict | None:
-    """The inspection profile's ambiguous result when it offers a choice.
+def inspection_choice(metadata: dict, file_name: str | None = None) -> dict | None:
+    """The inspection profile's result when the file is a review case.
 
     The scan asks only for a file the default profile found ambiguous
-    without a choice that nothing links.
+    without a choice that nothing links. The result is ambiguous with a
+    choice, or ``resolved`` when ``file_name`` anchors the file on its saver
+    and only that anchor resolves it (``saver_anchored``): the item then
+    offers the anchor's pair. None otherwise.
     """
     if inspection_review_version(metadata) is None:
         return None
-    resolution = resolve_inspection(metadata)
+    resolution = resolve_inspection(metadata, file_name)
+    if resolution.get("state") == "resolved":
+        return _anchor_of(metadata, file_name, resolution)
     if resolution.get("state") != "ambiguous" or not reviewable(resolution):
         return None
     return resolution
 
 
+def _anchor_of(metadata: Any, file_name: str | None, named: dict) -> dict | None:
+    """``named`` tagged ``anchor: "saver"`` when only the saver anchor
+    resolves the file, to a pair with a positive text to choose; else None.
+
+    ``named`` is ``metadata`` resolved with ``file_name``; the profile runs
+    once more without the name only when it is resolved and choosable.
+    """
+    if not file_name or named.get("state") != "resolved":
+        return None
+    try:
+        offered = choosable(_projection(named))
+    except (KeyError, TypeError):
+        return None
+    if not offered or resolve_inspection(metadata).get("state") == "resolved":
+        return None
+    return {**named, "anchor": SAVER_ANCHOR}
+
+
+def saver_anchored(metadata: Any, file_name: str | None) -> bool:
+    """True when the saver that wrote ``file_name`` resolves its branch (#390).
+
+    The inspection profile resolves ``metadata`` with the file's name, and
+    without it the file does not resolve: the anchor alone decides it. A file
+    the default profile left ambiguous is then queued with the anchor's pair
+    as its one candidate. No name, an unusable table, no matching saver,
+    several, or an unfinished walk anchor nothing (False). The profile
+    without the name runs only when the named run resolved.
+    """
+    return _anchored(metadata, file_name) is not None
+
+
+def _anchored(metadata: Any, file_name: str | None) -> dict | None:
+    """The anchored result ``saver_anchored`` describes, else None."""
+    if not file_name or not isinstance(metadata, dict):
+        return None
+    if inspection_review_version(metadata) is None:
+        return None
+    return _anchor_of(metadata, file_name, resolve_inspection(metadata, file_name))
+
+
 def item_profile(evidence: Any) -> str:
     """The profile an item's stored evidence was built with."""
     if isinstance(evidence, dict) and evidence.get("profile") == INSPECTION:
@@ -246,7 +319,9 @@ def record_ambiguous_file(
     """
     versions = item_versions(scanned[0] if scanned is not None else None)
     spelled_path, queued = _queued_spelling(db, root, relative, versions)
-    if queued is not None:
+    if queued is not None and not (
+        inspection and _default_item_to_anchor(db, queued, scanned, relative)
+    ):
         return _already_queued(db, *queued, versions)
     epoch = db.get_execution_epoch()
     try:
@@ -272,24 +347,24 @@ def record_ambiguous_file(
     if resolution is None or resolution.get("state") != "ambiguous":
         return "skipped"
     profile, version = DEFAULT, RESOLVER_VERSION
-    if not reviewable(resolution):
-        version = inspection_review_version(captured[0]) if inspection else None
-        if version is None:
+    named = _captured_inspection(captured[0], relative, scanned) if inspection else None
+    anchored = named is not None and named[0].get("anchor") == SAVER_ANCHOR
+    if anchored:
+        # Only the saver anchor resolves it (#390): offer that one pair.
+        profile, version = INSPECTION, named[1]
+        resolution = {
+            "state": "ambiguous",
+            "reason": ANCHOR_REASON,
+            "candidates": [named[0]],
+        }
+    elif not reviewable(resolution):
+        if named is None:
             return "untraced"
-        profile = INSPECTION
-        if (
-            scanned is not None
-            and scanned[1] is not None
-            and _same_records(captured[0], scanned[0])
-        ):
-            resolution = scanned[1]  # the same records: no second run
-        else:
-            # Heavier, so only for a file ambiguous with nothing to choose.
-            resolution = resolve_inspection(captured[0])
+        profile, (resolution, version) = INSPECTION, named
         if resolution.get("state") != "ambiguous" or not reviewable(resolution):
             return "untraced"
     evidence, count, limited = review_evidence(
-        resolution, profile, version, captured[0]
+        resolution, profile, version, captured[0], anchored=anchored
     )
     try:
         inserted = db.record_review_item(
@@ -313,6 +388,60 @@ def record_ambiguous_file(
     return "limited" if limited else "recorded"
 
 
+def _default_item_to_anchor(db, queued: tuple, scanned, relative: str) -> bool:
+    """True when the bytes' only pending item is a default item recorded
+    before #390 and the saver anchor alone resolves the scanned records.
+
+    The file is then captured again so the anchored item supersedes that
+    pending, undecided item. Decided bytes and bytes already queued under
+    the current inspection version are never re-read.
+    """
+    if scanned is None or scanned[0] is None:
+        return False
+    version = inspection_review_version(scanned[0])
+    if version is None or db.review_pending_at(*queued, version):
+        return False
+    if not db.review_pending_at(*queued, RESOLVER_VERSION):
+        return False
+    return _anchored(scanned[0], _file_name(relative)) is not None
+
+
+def _captured_inspection(metadata: dict, relative: str, scanned) -> tuple | None:
+    """``(resolution, item_version)`` of the captured records under the
+    inspection profile, anchored on the file's name; None when the table is
+    unusable.
+
+    A result tagged ``anchor: "saver"`` means the saver anchor alone
+    resolves the file (#390); any other resolved result is returned as
+    ``unresolved`` (no review case, as before). The scan's result is reused
+    when the capture holds the same records (``inspection_choice`` applies
+    the same rule).
+    """
+    version = inspection_review_version(metadata)
+    if version is None:
+        return None
+    if (
+        scanned is not None
+        and scanned[1] is not None
+        and _same_records(metadata, scanned[0])
+    ):
+        return scanned[1], version  # the same records: no second run
+    name = _file_name(relative)
+    resolution = resolve_inspection(metadata, name)
+    if resolution.get("state") == "resolved":
+        # Resolved without the anchor too: no review case, as before.
+        resolution = _anchor_of(metadata, name, resolution) or {
+            **resolution,
+            "state": "unresolved",
+        }
+    return resolution, version
+
+
+def _file_name(relative: str) -> str:
+    """The last part of a ``/``-separated relative path (the saved name)."""
+    return relative.rsplit("/", 1)[-1]
+
+
 def _already_queued(db, locator: str, size: int, mtime_ns: int, versions) -> str:
     """``duplicate`` while an existing item is pending, else ``decided``."""
     for version in versions:
@@ -354,14 +483,19 @@ def _queued_spelling(db, root, relative: str, versions) -> tuple[str, tuple | No
     return spelled, None
 
 
-def captured_review(store, capture, profile: str = DEFAULT) -> tuple[dict, dict] | None:
+def captured_review(
+    store, capture, profile: str = DEFAULT, file_name: str | None = None
+) -> tuple[dict, dict] | None:
     """``(metadata, resolution)`` of the capture's own bytes, else None.
 
     ``metadata`` is what the scan hands the ingester: prompt and workflow
     JSON-decoded and ``file_info`` (size, dimensions, format) from the
     capture. ``resolution`` comes from ``profile`` (``DEFAULT`` or
-    ``INSPECTION``). None when the capture is gone or carries no prompt
-    metadata.
+    ``INSPECTION``; the latter anchored on ``file_name``, the file's own
+    name, as the item was recorded). With ``file_name`` (Apply), a default
+    result its saver anchor alone resolves (``saver_anchored``) is that
+    anchored result, so only the anchored branch is still offered. None
+    when the capture is gone or carries no prompt metadata.
     """
     if store.acquire(capture.token) is None:
         return None
@@ -389,8 +523,15 @@ def captured_review(store, capture, profile: str = DEFAULT) -> tuple[dict, dict]
         info["format"] = read.format
     canonical["file_info"] = info
     if profile == INSPECTION:
-        return canonical, resolve_inspection(canonical)
-    return canonical, resolve_prompt_pair(canonical)
+        named = resolve_inspection(canonical, file_name)
+        return canonical, _anchor_of(canonical, file_name, named) or named
+    resolution = resolve_prompt_pair(canonical)
+    if file_name is not None and resolution.get("state") == "ambiguous":
+        anchored = _anchored(canonical, file_name)
+        if anchored is not None:
+            # Its saver names its branch (#390): only that one is offered.
+            return canonical, anchored
+    return canonical, resolution
 
 
 def reviewable(resolution: Any) -> bool:
@@ -416,12 +557,14 @@ def review_evidence(
     profile: str = DEFAULT,
     version: int | None = None,
     metadata: Any = None,
+    anchored: bool = False,
 ) -> tuple[dict, int, bool]:
     """``(evidence, candidate_count, limited)`` for one ambiguous result.
 
     Inspection evidence also names its ``profile`` and ``semantics``, the
     table fingerprint of ``metadata``'s graph it was resolved with;
-    ``version`` is the item's version.
+    ``version`` is the item's version. ``anchored`` evidence (the saver
+    anchor's one pair, #390) also names ``anchor: "saver"``.
     """
     base = {
         "resolver_version": RESOLVER_VERSION if version is None else version,
@@ -430,6 +573,8 @@ def review_evidence(
     if profile == INSPECTION:
         base["profile"] = INSPECTION
         base["semantics"] = semantics_version(metadata)
+    if anchored:
+        base["anchor"] = SAVER_ANCHOR
     candidates: list[dict] = []
     seen: set[str] = set()
     total = 0

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/system-diagnostics.mjs

@@ -225,6 +225,23 @@ export function clearSentence(files) {
     return `Empties ${ACTIVE_LOG}, deletes ${number(rotated.length)} rotated ${noun(rotated.length, 'copy', 'copies')} (${sizeText(size)}) ${tail}`;
 }
 
+/**
+ * #389: the `/logs/truncate` body for the files a confirmation showed. Their active and rotated rows go back with the
+ * server's `identity`, so the server deletes only those and refuses (409) when any changed meanwhile. A list that could
+ * not be read (`null`, the general sentence) sends no list: the server then clears what is there, as the sentence says.
+ */
+export function clearBody(files) {
+    if (!Array.isArray(files)) return {};
+    const expected = files.filter(row => (row?.kind === 'active' || row?.kind === 'rotated') && typeof row.filename === 'string'
+        && typeof row.identity === 'string').map(row => ({ name: row.filename, identity: row.identity }));
+    return { expected };
+}
+
+/** A refused clear (409) says why in the server's words; any other failure the general line. */
+export function clearFailureText(error) {
+    return error?.status === 409 && typeof error.message === 'string' && error.message ? error.message : COPY.clearFailed;
+}
+
 /** CUJ 7: the toast for `/logs/truncate`'s `results` ({truncated: [names], errors: [messages]}). */
 export function clearToast(results) {
     const truncated = Array.isArray(results?.truncated) ? results.truncated : [];
@@ -469,13 +486,14 @@ function logsSurface({ api, activity, scope, drafts, navigate, notify, latestOnl
     const list = el('div', undefined, 'operation-list');
     let listed = null;
     // The confirmation names what Clear deletes now: the files are read again before it opens (a read that fails
-    // gives the general sentence, never an old count).
+    // gives the general sentence, never an old count). Clear sends exactly the files the dialog was built from (#389).
     const clear = operationRow({ scope, title: 'Clear all logs', description: COPY.clearDescription, label: 'Clear all logs…',
         danger: true, onClick: button => pressing(button, async () => {
             await readFiles();
             if (scope.signal.aborted) return;
-            confirmDialog({ scope, trigger: button, title: 'Clear all logs', sentence: clearSentence(listed), label: 'Clear',
-                danger: true, onConfirm: clearLogs });
+            const shown = listed;
+            confirmDialog({ scope, trigger: button, title: 'Clear all logs', sentence: clearSentence(shown), label: 'Clear',
+                danger: true, onConfirm: () => clearLogs(shown) });
         }) });
     files.append(el('summary', 'Log files & retention'), line, review, retention, list, clear);
     const runtime = runtimeRows({ scope, drafts });
@@ -562,12 +580,12 @@ function logsSurface({ api, activity, scope, drafts, navigate, notify, latestOnl
         readConfig();
     }
 
-    async function clearLogs() {
+    async function clearLogs(shown) {
         let answer;
         try {
-            answer = await api.post(TRUNCATE_URL, {});
-        } catch {
-            if (!scope.signal.aborted) notify(COPY.clearFailed);
+            answer = await api.post(TRUNCATE_URL, clearBody(shown));
+        } catch (error) {
+            if (!scope.signal.aborted) notify(clearFailureText(error));
         }
         if (scope.signal.aborted) return;
         if (answer) notify(clearToast(answer.results));