Merge remote-tracking branch 'origin/main' into feat/329-s12-pr1

# Conflicts:
#	__init__.py

.github/branch-protection.json

@@ -1,6 +1,6 @@
 {
   "required_status_checks": {
-    "strict": false,
+    "strict": true,
     "contexts": [
       "py3.12 on ubuntu-latest",
       "py3.13 on ubuntu-latest",

.github/workflows/code-quality.yml

@@ -4,7 +4,6 @@ permissions:
   contents: read
 
 on:
-  merge_group:
   push:
     branches: [main]
   pull_request:

.github/workflows/test.yml

@@ -4,7 +4,6 @@ permissions:
   contents: read
 
 on:
-  merge_group:
   pull_request:
     branches: [main]
   push:
@@ -64,18 +63,16 @@ jobs:
   comfyui:
     name: real ComfyUI compatibility
     runs-on: ubuntu-latest
-    # Chromium and its system libraries come from the pinned Playwright image (tag = @playwright/test in
-    # tools/vendor/package-lock.json; tests/vendor/browser-image.test.mjs checks it). The runner install
-    # pulled them from Ubuntu's apt mirror on every run, and three runs (#349, #378, #382) crawled at
-    # ~26 kB/s for 11-12 minutes until the job hit its limit after every test had passed.
-    container:
-      image: mcr.microsoft.com/playwright:v1.64.1-noble
-      options: --user 1001
     timeout-minutes: 24
     steps:
       - uses: actions/checkout@v7
         with:
           persist-credentials: false
+      - uses: actions/setup-node@v7
+        with:
+          node-version: "24"
+          cache: npm
+          cache-dependency-path: tools/vendor/package-lock.json
       - name: Read the reviewed runtime pins
         id: runtime
         run: |
@@ -100,42 +97,21 @@ jobs:
           ref: ${{ steps.runtime.outputs.kikoRevision }}
           path: .ci-kikotools
           persist-credentials: false
-      # The image's own Python and Node 24: setup-python/setup-node install the runner's toolcache builds,
-      # whose entry points name host paths that do not exist in the container. The image has no ensurepip
-      # and the job has no root, so a checksum-pinned pip wheel bootstraps a user-space venv. The image
-      # Python must be the reviewed host Python, so a pin or image bump that disagrees fails here.
-      - name: Python environment on the reviewed host Python
-        env:
-          PIP_WHEEL: https://files.pythonhosted.org/packages/f3/6e/1736e5b4ae2b778ef2f81c47d797de9f891d4d8acb047a24ca37a60294dd/pip-26.2.1-py3-none-any.whl
-          PIP_SHA256: 71138adf1f4ca900cdb7d289c21b7494329f2332b6d85f0e1c42108c0384ed3e
-          PINNED_PYTHON: ${{ steps.runtime.outputs.pythonVersion }}
-        run: |
-          image="$(python3 -c 'import sys; print("%d.%d" % sys.version_info[:2])')"
-          [ "$image" = "$PINNED_PYTHON" ] || { echo "::error::image Python $image is not the pinned $PINNED_PYTHON"; exit 1; }
-          venv="$RUNNER_TEMP/venv" wheel="$RUNNER_TEMP/pip-26.2.1-py3-none-any.whl"
-          python3 -m venv --without-pip "$venv"
-          curl -fsSL --retry 3 -o "$wheel" "$PIP_WHEEL"
-          echo "$PIP_SHA256  $wheel" | sha256sum -c -
-          "$venv/bin/python" "$wheel/pip" install --no-index --no-cache-dir "$wheel"
-          echo "$venv/bin" >> "$GITHUB_PATH"
-      - run: python --version && node --version && node --version | grep -q '^v24\.'
-      # pip's download cache (torch alone is ~200 MB); inside the container ~ is the job's HOME (/github/home).
-      - name: Cache pip downloads
-        uses: actions/cache@v6
+      - uses: actions/setup-python@v7
         with:
-          path: ~/.cache/pip
-          key: pip-${{ runner.os }}-playwright-v1.64.1-${{ hashFiles('requirements-dev.txt', '.ci-comfyui/requirements.txt', 'tools/vendor/comfyui-runtime.json') }}
+          python-version: ${{ steps.runtime.outputs.pythonVersion }}
+          cache: pip
+          cache-dependency-path: |
+            requirements-dev.txt
+            .ci-comfyui/requirements.txt
+            tools/vendor/comfyui-runtime.json
       - name: Install CPU runtime and test dependencies
         run: |
+          python -m pip install --upgrade pip
           pip install "torch==${{ steps.runtime.outputs.torchVersion }}" "torchvision==${{ steps.runtime.outputs.torchvisionVersion }}" --index-url https://download.pytorch.org/whl/cpu
           pip install -r .ci-comfyui/requirements.txt -r requirements-dev.txt "torch==${{ steps.runtime.outputs.torchVersion }}" "torchvision==${{ steps.runtime.outputs.torchvisionVersion }}"
-      # npm's download cache, keyed on the lockfile. node_modules itself is still built fresh by npm ci.
-      - name: Cache npm downloads
-        uses: actions/cache@v6
-        with:
-          path: ~/.npm
-          key: npm-${{ runner.os }}-playwright-v1.64.1-${{ hashFiles('tools/vendor/package-lock.json') }}
       - run: npm ci --prefix tools/vendor --ignore-scripts
+      - run: npx --prefix tools/vendor playwright install --with-deps chromium
       - name: Classic and Nodes 2.0 startup and compatibility
         env:
           PM_COMFY_ROOT: ${{ github.workspace }}/.ci-comfyui

.github/workflows/viewerjs-check.yml

@@ -1,7 +1,6 @@
 name: Viewer.js vendor and compatibility
 
 on:
-  merge_group:
   pull_request:
     branches: [main]
   push:

TESTING.md

@@ -282,12 +282,11 @@ PromptManager's own background threads stop through one package teardown,
 works against one 10 s deadline. Each step gets the smaller of its own bound
 and the time left. The steps, in order:
 
-1. Refuse new previews (`refuse_auto_thumbnails()`), then detach the
-   monitor's callbacks (Needs review recorder, preview queue, Images listing
-   listener).
-2. `ImageMonitor.stop_monitoring` works against one 5 s deadline. Closing the
-   image handler joins the image worker; watchdog's `observer.stop()` (not
-   bounded here) and the observer join get what is left, which can be 0 s.
+1. Detach the monitor's callbacks (Needs review recorder, preview queue,
+   Images listing listener).
+2. `ImageMonitor.stop_monitoring` closes the image handler, which joins the
+   image worker (5 s). It then calls watchdog's `observer.stop()`, which is
+   not bounded here, and joins the observer (5 s).
 3. `close_auto_thumbnails(final=True)` stops the preview worker (30 s) and
    kills ffmpeg.
 4. `PromptTracker.shutdown` stops the prompt cleanup thread (2 s).
@@ -296,11 +295,7 @@ A failing step, or a second Ctrl+C, is logged with its traceback and does not
 skip the later steps. A clean run logs one INFO line to PromptManager's log:
 `Background work stopped in 0.4 s: monitor=ok thumbnails=ok tracker=ok`. If
 any part is still running, one WARNING goes to the console:
-`Background work not stopped in time (10.0 s): monitor`. A Ctrl+C that lands
-between two steps (each step catches its own) is re-raised after the result is
-stored, with the parts not reached False, and the `atexit` callback is
-removed: the later steps are skipped, no summary line is logged, and a later
-call returns that result and runs nothing.
+`Background work not stopped in time (10.0 s): monitor`.
 
 ComfyUI never runs aiohttp's `on_cleanup` hooks, because it never calls
 `runner.cleanup()`. So `atexit` is the live host's path. It runs on Ctrl+C
@@ -3496,9 +3491,7 @@ workflow (README changelog).
 ### LoRA reference import contract (#221, #249 System S10)
 
 The LoRA import is staged (`py/lora_import.py`): discover the LoraManager
-`*.metadata.json` files and fingerprint every scanned one, kept or skipped (path, size,
-modification time and whether it was kept, #362; a skipped sidecar added, removed or edited
-after the review refuses Start with `plan-changed`, `test_start_refused_when_a_skipped_sidecar_changed`); stage each
+`*.metadata.json` files and fingerprint them (path, size, modification time); stage each
 record (CivitAI example images go into the image cache first, best effort, counted in
 `downloaded` / `download_errors`; each image is written to a temp file beside its cache
 entry and moved in place, so a half-written image is never seen as cached); the files are
@@ -3564,67 +3557,25 @@ the kept, readable LoRAs and `skipped` the rest. Media per LoRA, without repeats
 preview (`preview_url`, else LoraManager's `PREVIEW_EXTENSIONS` order), then every file in
 `<example-images root>[/<library>]/<sha256>/` in natural order; a leading video yields to the
 first still. A file counts only with a servable extension and inside the guarded media
-roots. Only an absolute `preview_url` is used; a relative one is skipped with a DEBUG line
-naming the sidecar. A sidecar that is not valid UTF-8 JSON is skipped like any unreadable
-one (#362). CivitAI downloads run only for a LoRA without local media; a record's
-images are linked in reverse order, so its first image (the cover) is the newest link and
-the card image. The folder guard keeps a folder from the settings, and each of ComfyUI's
-default LoRA folders (`models/loras`, `extra_model_paths.yaml`, `folder_paths`), only when
-it exists and is neither a filesystem root (`/`, `C:\`, a UNC share root) nor the home
-folder or a folder that contains it (realpath + `normcase`); a refused default folder that
-exists is listed in the layout's `refused_defaults` and logged as one warning each time the
-refused set changes (`RefusedDefaultsTests`). An unusable layout (no install, unreadable /
-non-JSON / non-object / oversized settings, no `platformdirs`, a missing or refused
-centralized sidecar root, "the sidecar folder is a filesystem root, the home folder or a
-folder that contains it") gives a review without a token ("LoraManager settings: not
-usable: <reason>. Nothing was changed."), a legacy scan error with that reason, the default
-LoRA roots for serving (no example-images root), and trigger words from the default roots
-in alongside mode. A folder discovery cannot list answers the review with 500, code
-`folder-unreadable` ("A LoRA folder could not be listed (...). Nothing was changed.").
-`TriggerWordCache.load` never raises: an `OSError` is a warning, anything else is logged
-with its traceback (#362), both keep the previous map, and a running node retries a failed
-load only after 60 s (`TRIGGER_RETRY_SECONDS`). `tests/run_suite.py` and `tests/browser/server.py` point
+roots. CivitAI downloads run only for a LoRA without local media. The folder guard keeps a
+settings folder only when it exists and is neither a filesystem root (`/`, `C:\`, a UNC
+share root) nor the home folder or one of its ancestors (realpath + `normcase`). An
+unusable layout (no install, unreadable / non-JSON / non-object / oversized settings, no
+`platformdirs`, a missing or refused centralized sidecar root) gives a review without a
+token ("LoraManager settings: not usable: <reason>. Nothing was changed."), a legacy scan
+error with that reason, the default LoRA roots for serving, and trigger words from the
+default roots in alongside mode. `tests/run_suite.py` and `tests/browser/server.py` point
 `LORA_MANAGER_SETTINGS_DIR` at a disposable folder, so no test reads the developer's real
 LoraManager settings; tests that call `load_layout` inject their own folder too.
 
-**Linked-only gate and the layout cache (#362).** While the integration is on, the LoRA
-allow-list (`ImagesMixin._lora_image_dirs`) is the layout's `media_roots` plus the CivitAI
-download cache, and `_media_roots(outputs)` appends it after the output folders for Library
-previews, the Library prompt detail, the per-prompt images page, `serve_output_image` for a
-root past the outputs, `inspect_image` and `reveal_image`. Under such a root only a file the
-database records as a linked image is served, inspected or revealed (`_linked_only_error`,
-by its given and its real path); anything else answers 403. Library responses mark that
-media `linked_only`: no preview file, no favorite, no Tag button. The allow-list's layout
-comes from `py/lora_layout_cache.py` (`tests/test_lora_layout_cache.py`, which injects the
-clocks, `stat` and `ntpath` and never sleeps): one entry keyed by the stripped configured
-path, reused while the plugin settings file's and the settings file's `(st_mtime_ns,
-st_size)` (or "missing") are unchanged, the install and every media root still resolve to
-themselves (a folder swapped for a symlink reloads and the guard runs again), and the entry
-is under `MAX_AGE_SECONDS` (30 s, monotonic clock) old. A load is stored only when no
-watched file was modified within `RACY_SECONDS` (3 s) of the load (git's racy rule: a write
-during the load, LoraManager's in-place write, FAT/HFS+ coarse times), every watched file
-could be checked (any `stat` error but not-found, and an offline Windows share, is "cannot
-validate"), the settings were readable, and no `clear()` ran meanwhile (generation
-counter). Hits take no lock; concurrent misses each load. A finished import (System job or
-legacy scan, both through `replace_lora_imports`) clears it. The integration-off check runs
-before it, and the configured-path rule (a configured path that is not an install under
-`custom_nodes` adds no folders) on its result; a hit also needs the same home folder, which
-the guard compares against. `serve_image` builds the allow-list in the
-executor. The cache writes one INFO line, "LoraManager layout changed: <source> settings,
-<mode> storage, <n> media folders", when a stored layout differs from the last one logged.
-The review, Start, the legacy scan and the trigger cache read the layout fresh. Image route
-tests clear the per-process cache in `ImageRouteCoverageCase.setUp`; `write_settings` in
-`TestLoraManagerMediaServing` back-dates its file 10 s so the first load is stored.
-
 `tests/browser/workspace_integrations_fixture.py` (`/__test__/integrations/*`) makes a fake
 LoraManager install and LoRA folder under the server's temp root, holds or fails the
 import, saves a blocking pair and answers CivitAI downloads with a stub opener, so no
 test reaches the network.
 
 ```sh
-python3 -m unittest tests.test_lora_layout_cache tests.test_lora_manager_layout \
-  tests.test_lora_import tests.test_api_lora tests.test_lora_database tests.test_lora_utils \
-  tests.test_api_images -v
+python3 -m unittest tests.test_lora_manager_layout tests.test_lora_import tests.test_api_lora \
+  tests.test_lora_database tests.test_lora_utils -v
 node --test tests/js/workspace-system-integrations.test.js tests/js/prompt-bridge-adapter.test.js
 npm run --prefix tools/vendor test:workflows -- workspace-system-integrations.spec.mjs \
   system-integrations-concept.spec.mjs \

__init__.py

@@ -23,24 +23,20 @@ Features:
 """
 
 import atexit
-import re
 import threading
 import time
 from pathlib import Path
 
 
 def get_version():
-    """Parse version from pyproject.toml"""
+    """Parse version from pyproject.toml (``utils/version.py``)."""
+    from .utils.version import project_version
+
     try:
-        pyproject_path = Path(__file__).parent / "pyproject.toml"
-        if pyproject_path.exists():
-            content = pyproject_path.read_text()
-            match = re.search(r'version\s*=\s*["\']([^"\']+)["\']', content)
-            if match:
-                return match.group(1)
+        root = Path(__file__).parent
     except Exception:
-        pass
-    return "unknown"
+        return "unknown"
+    return project_version(root)
 
 
 from .prompt_manager import (  # noqa: E402
@@ -268,89 +264,78 @@ def _stop_background_work():
 
     Returns ``{"monitor", "thumbnails", "tracker"}`` booleans (True: it
     stopped), leaving out parts that never started. Later calls return a
-    copy of the first result. A Ctrl+C inside a step is logged and the
-    step is False; one between steps is re-raised after the result (parts
-    not reached are False) is stored and the ``atexit`` entry removed.
+    copy of the first result.
     """
     global _background_work_result
     with _background_work_lock:
         if _background_work_result is not None:
             return dict(_background_work_result)
+        logger = _teardown_logger()
+        began = _clock()
+        deadline = began + TEARDOWN_SECONDS
+
+        def left(own_bound):
+            return max(0.0, min(own_bound, deadline - _clock()))
+
         monitor, thumbnails, tracker = (
             _global_image_monitor,
             api,
             _global_prompt_tracker,
         )
-        # A part whose global is set has started; False until its step says
-        # otherwise, so an interrupt between steps leaves an honest result.
-        started = (
-            ("monitor", monitor),
-            ("thumbnails", thumbnails),
-            ("tracker", tracker),
-        )
-        result = {name: False for name, part in started if part is not None}
-        try:
-            logger = _teardown_logger()
-            began = _clock()
-            deadline = began + TEARDOWN_SECONDS
+        result = {}
+        if monitor is not None:
 
-            def left(own_bound):
-                return max(0.0, min(own_bound, deadline - _clock()))
+            def stop_monitor():
+                from .utils.image_monitor import OBSERVER_JOIN_SECONDS
 
-            if monitor is not None:
+                # Callbacks first: an image in hand queues no preview into a
+                # closing preview worker. The worker may have read the queue
+                # callback already, so the API refuses new previews too (at
+                # exit, Python 3.12+ refuses to start their worker thread).
+                if thumbnails is not None:
+                    thumbnails.refuse_auto_thumbnails()
+                monitor.set_review_recorder(None)
+                monitor.set_thumbnail_queue(None)
+                monitor.set_output_listener(None)
+                return monitor.stop_monitoring(left(OBSERVER_JOIN_SECONDS))
 
-                def stop_monitor():
-                    from .utils.image_monitor import OBSERVER_JOIN_SECONDS
+            result["monitor"] = _teardown_step(
+                "the image monitor", stop_monitor, logger
+            )
+        if thumbnails is not None:
 
-                    # Callbacks first: an image in hand queues no preview into a
-                    # closing preview worker. The worker may have read the queue
-                    # callback already, so the API refuses new previews too (at
-                    # exit, Python 3.12+ refuses to start their worker thread).
-                    if thumbnails is not None:
-                        thumbnails.refuse_auto_thumbnails()
-                    monitor.set_review_recorder(None)
-                    monitor.set_thumbnail_queue(None)
-                    monitor.set_output_listener(None)
-                    return monitor.stop_monitoring(left(OBSERVER_JOIN_SECONDS))
+            def close_previews():
+                from .py.api.thumbnail_auto import CLOSE_WAIT_SECONDS
 
-                result["monitor"] = _teardown_step(
-                    "the image monitor", stop_monitor, logger
+                return thumbnails.close_auto_thumbnails(
+                    final=True, timeout=left(CLOSE_WAIT_SECONDS)
                 )
-            if thumbnails is not None:
 
-                def close_previews():
-                    from .py.api.thumbnail_auto import CLOSE_WAIT_SECONDS
+            result["thumbnails"] = _teardown_step(
+                "the preview worker", close_previews, logger
+            )
+        if tracker is not None:
 
-                    return thumbnails.close_auto_thumbnails(
-                        final=True, timeout=left(CLOSE_WAIT_SECONDS)
-                    )
+            def stop_tracker():
+                from .utils.prompt_tracker import TRACKER_JOIN_SECONDS
 
-                result["thumbnails"] = _teardown_step(
-                    "the preview worker", close_previews, logger
-                )
-            if tracker is not None:
+                return tracker.shutdown(left(TRACKER_JOIN_SECONDS))
 
-                def stop_tracker():
-                    from .utils.prompt_tracker import TRACKER_JOIN_SECONDS
-
-                    return tracker.shutdown(left(TRACKER_JOIN_SECONDS))
-
-                result["tracker"] = _teardown_step(
-                    "the prompt cleanup thread", stop_tracker, logger
-                )
-            elapsed = _clock() - began
-            failed = [name for name, ok in result.items() if not ok]
-            if failed:
-                logger.warning(
-                    f"Background work not stopped in time ({elapsed:.1f} s): "
-                    + ", ".join(failed)
-                )
-            else:
-                parts = " ".join(f"{name}=ok" for name in result)
-                logger.info(f"Background work stopped in {elapsed:.1f} s: {parts}")
-        finally:
-            _background_work_result = result
-            atexit.unregister(_stop_background_work)
+            result["tracker"] = _teardown_step(
+                "the prompt cleanup thread", stop_tracker, logger
+            )
+        elapsed = _clock() - began
+        failed = [name for name, ok in result.items() if not ok]
+        if failed:
+            logger.warning(
+                f"Background work not stopped in time ({elapsed:.1f} s): "
+                + ", ".join(failed)
+            )
+        else:
+            parts = " ".join(f"{name}=ok" for name in result)
+            logger.info(f"Background work stopped in {elapsed:.1f} s: {parts}")
+        _background_work_result = result
+        atexit.unregister(_stop_background_work)
         return dict(result)
 
 

py/api/__init__.py

@@ -5,6 +5,8 @@ Split into domain-specific mixins for maintainability:
 - ImageRoutesMixin: gallery, thumbnails, image serving and linking
 - AdminRoutesMixin: duplicates, stats, settings, diagnostics, maintenance, backup, scan
 - LoggingRoutesMixin: log management endpoints
+- LogCleanupMixin: the reviewed rotated-log cleanup (a maintenance plan)
+- DiagnosticsReportMixin: the shareable diagnostic report, without secrets or paths
 - AutotagRoutesMixin: auto-tagging model management and tagging
 - JobRoutesMixin: the session job registry behind Activity (py/api/jobs.py)
 - ImageTagJobsMixin: bulk Add tags / Auto-tag as image-tags jobs
@@ -65,6 +67,8 @@ from .duplicate_files import DuplicateFilesMixin
 from .maintenance_plans import PlanStore
 from .jobs import JobRegistry
 from .logging_routes import LoggingRoutesMixin
+from .log_cleanup import LogCleanupMixin
+from .diagnostics_report import DiagnosticsReportMixin
 from .autotag_routes import AutotagRoutesMixin
 from .lora_integration import LoraIntegrationMixin
 from .lora_import_routes import LoraImportRoutesMixin
@@ -323,6 +327,8 @@ class PromptManagerAPI(
     WorkflowStorageRoutesMixin,
     DuplicateFilesMixin,
     LoggingRoutesMixin,
+    LogCleanupMixin,
+    DiagnosticsReportMixin,
     AutotagRoutesMixin,
     LoraIntegrationMixin,
     LoraImportRoutesMixin,
@@ -678,6 +684,7 @@ class PromptManagerAPI(
         self._register_settings_routes(routes)
         self._register_job_routes(routes)
         self._register_system_routes(routes)
+        self._register_diagnostics_report_routes(routes)
         self._register_library_location_routes(routes)
         self._register_maintenance_routes(routes)
         self._register_backup_restore_routes(routes)

py/api/admin.py

@@ -489,9 +489,29 @@ def _diagnose_dependencies():
     }
 
 
-def _diagnose_output_dirs(configured_dirs):
+def legacy_checks(db_path, public_path=None):
+    """The four checks of ``GET /diagnostics`` (blocking).
+
+    Shared with the diagnostic report (``py/api/diagnostics_report.py``),
+    which passes its own ``public_path`` so a folder outside ComfyUI is never
+    named.
+    """
+    from ..config import GalleryConfig
+
+    return {
+        "database": _diagnose_database(db_path),
+        "dependencies": _diagnose_dependencies(),
+        "comfyui_output": _diagnose_output_dirs(
+            GalleryConfig.MONITORING_DIRECTORIES, public_path
+        ),
+        "image_monitor": _diagnose_image_monitor(public_path),
+    }
+
+
+def _diagnose_output_dirs(configured_dirs, public_path=None):
     """Configured gallery roots, ComfyUI's output dir, or relative fallbacks."""
     _public_path, _ = _public_helpers()
+    _public_path = public_path or _public_path
     candidates = list(configured_dirs)
     try:
         import folder_paths
@@ -525,9 +545,10 @@ def _diagnose_output_dirs(configured_dirs):
     }
 
 
-def _diagnose_image_monitor():
+def _diagnose_image_monitor(public_path=None):
     """Status of the running image monitor, with public directory names."""
     _public_path, _public_error = _public_helpers()
+    _public_path = public_path or _public_path
     try:
         monitor = _current_image_monitor()
         if monitor is None:
@@ -1055,16 +1076,7 @@ class AdminRoutesMixin:
     async def run_diagnostics(self, request):
         """Run comprehensive system diagnostics and health checks."""
         try:
-            from ..config import GalleryConfig
-
-            results = {
-                "database": _diagnose_database(self.db.model.db_path),
-                "dependencies": _diagnose_dependencies(),
-                "comfyui_output": _diagnose_output_dirs(
-                    GalleryConfig.MONITORING_DIRECTORIES
-                ),
-                "image_monitor": _diagnose_image_monitor(),
-            }
+            results = legacy_checks(self.db.model.db_path)
             return web.json_response({"success": True, "diagnostics": results})
 
         except Exception as e:

py/api/diagnostics_report.py

@@ -0,0 +1,289 @@
+"""The shareable diagnostic report (#329 System S12).
+
+``GET /prompt_manager/system/diagnostics`` answers a report built on the
+server from an explicit list of fields: the project and Viewer.js versions,
+Python and the platform, the four checks of the legacy ``GET /diagnostics``
+(one helper, ``admin.legacy_checks``), the database's public info, source
+counts, the log configuration and statistics, the last health summary and
+the configuration keys that only ``config.json`` holds. It never copies
+``PromptManagerConfig.get_config()``: no path leaf is listed, and the CivitAI
+key is reported only as ``has_civitai_api_key``.
+
+Every path in the report goes through ``report_path``: relative to the
+closest public anchor that is not a filesystem root, else ``"outside
+ComfyUI"``, never a basename that could be a user name. As a backstop,
+``scrub`` replaces any string that still looks like an absolute path, holds
+the saved key or the home folder (with or without its root), and counts each
+replacement in ``redacted`` (0 for a correct builder).
+
+The report opens the database, so it takes a registry lease: it is refused
+(409 ``busy``) while an ``exclusive`` job (a restore) runs, and a restore
+waits for it. It is built off the event loop.
+"""
+
+import datetime
+import json
+import ntpath
+import os
+import platform
+import posixpath
+import re
+from pathlib import Path, PurePosixPath, PureWindowsPath
+
+from aiohttp import web
+
+from . import admin as _admin
+from .job_routes import busy_response
+from .jobs import JobBusy
+from .system import health_summary
+
+try:
+    from ...utils.version import project_version
+except ImportError:  # loaded as the top-level ``py`` package (tests)
+    from utils.version import project_version
+
+DIAGNOSTICS_LEASE = "Diagnostics"
+_ROOT = Path(__file__).resolve().parents[2]
+VIEWERJS_PROVENANCE = str(_ROOT / "web" / "lib" / "viewerjs" / "provenance.json")
+OUTSIDE = "outside ComfyUI"
+REMOVED = "[removed]"
+# The CivitAI key is searched for only when it is long enough to be a key.
+MIN_KEY_LENGTH = 8
+_WINDOWS_ABSOLUTE = re.compile(r"^[A-Za-z]:[\\/]")
+
+
+def anchored_path(canonical, anchors, home, pathmod=os.path):
+    """``canonical`` relative to the closest usable anchor, else ``outside ComfyUI``.
+
+    All three are already canonical (``normcase``) in ``pathmod``'s spelling.
+    A filesystem root is never usable, nor is a strict ancestor of ``home``
+    (``/home``, ``C:\\Users``), whose relative paths would start with a user
+    name; so a path inside the home folder needs an anchor at or below it.
+    """
+    flavor = PureWindowsPath if pathmod is ntpath else PurePosixPath
+    target = flavor(canonical)
+    home_path = flavor(home) if home else None
+    usable = []
+    for anchor in map(flavor, anchors):
+        if anchor.parent == anchor:
+            continue
+        if home_path is not None and home_path != anchor:
+            if home_path.is_relative_to(anchor):
+                continue
+        if target.is_relative_to(anchor):
+            usable.append(anchor)
+    if not usable:
+        return OUTSIDE
+    anchor = max(usable, key=lambda a: len(a.parts))
+    return target.relative_to(anchor).as_posix() or "."
+
+
+def _canonical(path):
+    return os.path.normcase(os.path.realpath(str(path)))
+
+
+def report_path(path):
+    """``path`` relative to a public anchor (POSIX style), else ``outside ComfyUI``."""
+    if not path:
+        return ""
+    from ..config import GalleryConfig
+
+    try:
+        canonical = _canonical(path)
+        anchors = [_canonical(anchor) for anchor in GalleryConfig.path_anchors()]
+        home = os.path.expanduser("~")
+        home = _canonical(home) if home and home != "~" else None
+    except (OSError, ValueError):
+        return OUTSIDE
+    return anchored_path(canonical, anchors, home)
+
+
+def _looks_absolute(text):
+    return (
+        posixpath.isabs(text)
+        or ntpath.isabs(text)
+        or _WINDOWS_ABSOLUTE.match(text) is not None
+        or text.startswith("\\\\")
+    )
+
+
+# The home folder is matched only as whole path parts, never inside a word.
+_PART = r"(?<![\w.-]){}(?![\w.-])"
+MIN_HOME_NAME = 3
+
+
+def _home_patterns():
+    """``(pattern, paths_only)`` for the home folder: in full, rootless, its name.
+
+    Accepted residual risk: only this process's own home folder is known, so
+    another user's home (an output folder under /home/<them> while ComfyUI
+    runs as a service account) can still be named by a report path.
+    """
+    home = os.path.expanduser("~")
+    if not home or home == "~" or len(home) < 4 or home in ("/", "\\"):
+        return ()
+    spellings = {home, home.replace("\\", "/")}
+    relative = os.path.splitdrive(home)[1].strip("\\/")
+    for spelling in (relative, relative.replace("\\", "/")):
+        if "/" in spelling or "\\" in spelling:
+            spellings.add(spelling)
+    patterns = [
+        (re.compile(_PART.format(re.escape(spelling)), re.IGNORECASE), False)
+        for spelling in spellings
+    ]
+    name = re.split(r"[\\/]", relative)[-1] if relative else ""
+    if len(name) >= MIN_HOME_NAME:
+        # The bare name only inside a path ("info/x"), not a word ("INFO").
+        patterns.append(
+            (re.compile(_PART.format(re.escape(name)), re.IGNORECASE), True)
+        )
+    return tuple(patterns)
+
+
+def scrub(report, key):
+    """``report`` with leaked paths and the key replaced, counted in ``redacted``."""
+    key = key if isinstance(key, str) and len(key) >= MIN_KEY_LENGTH else None
+    homes = _home_patterns()
+    count = 0
+
+    def clean(value):
+        nonlocal count
+        if isinstance(value, str):
+            if (
+                _looks_absolute(value)
+                or (key and key in value)
+                or any(
+                    pattern.search(value)
+                    for pattern, paths_only in homes
+                    if not paths_only or "/" in value or "\\" in value
+                )
+            ):
+                count += 1
+                return REMOVED
+            return value
+        if isinstance(value, dict):
+            return {name: clean(item) for name, item in value.items()}
+        if isinstance(value, (list, tuple)):
+            return [clean(item) for item in value]
+        return value
+
+    cleaned = clean(report)
+    cleaned["redacted"] = int(cleaned.get("redacted") or 0) + count
+    return cleaned
+
+
+def viewerjs_version(path=None):
+    """The pinned Viewer.js version from its provenance file, or ``unknown``."""
+    try:
+        with open(path or VIEWERJS_PROVENANCE, encoding="utf-8") as handle:
+            version = json.load(handle).get("version")
+    except (OSError, ValueError, AttributeError):
+        return "unknown"
+    return version if isinstance(version, str) and version else "unknown"
+
+
+def public_config():
+    """The configuration keys only ``config.json`` holds, from an explicit list."""
+    from ..config import GalleryConfig, IntegrationConfig, PromptManagerConfig
+
+    return {
+        "database": {
+            "enable_duplicate_detection": (
+                PromptManagerConfig.ENABLE_DUPLICATE_DETECTION
+            ),
+            "enable_auto_save": PromptManagerConfig.ENABLE_AUTO_SAVE,
+        },
+        "web_ui": {
+            "result_timeout": PromptManagerConfig.RESULT_TIMEOUT,
+            "show_test_button": PromptManagerConfig.SHOW_TEST_BUTTON,
+            "webui_display_mode": PromptManagerConfig.WEBUI_DISPLAY_MODE,
+            "infinite_scroll": PromptManagerConfig.INFINITE_SCROLL,
+        },
+        "performance": {
+            "max_search_results": PromptManagerConfig.MAX_SEARCH_RESULTS,
+            "enable_fuzzy_search": PromptManagerConfig.ENABLE_FUZZY_SEARCH,
+            "auto_backup_interval": PromptManagerConfig.AUTO_BACKUP_INTERVAL,
+            "worker_threads": PromptManagerConfig.WORKER_THREADS,
+        },
+        "gallery": {
+            "processing_delay": GalleryConfig.PROCESSING_DELAY,
+            "prompt_timeout": GalleryConfig.PROMPT_TIMEOUT,
+            "cleanup_interval": GalleryConfig.CLEANUP_INTERVAL,
+            "auto_cleanup_missing_files": GalleryConfig.AUTO_CLEANUP_MISSING_FILES,
+            "max_image_age_days": GalleryConfig.MAX_IMAGE_AGE_DAYS,
+            "max_concurrent_processing": GalleryConfig.MAX_CONCURRENT_PROCESSING,
+            "metadata_extraction_timeout": GalleryConfig.METADATA_EXTRACTION_TIMEOUT,
+        },
+        "integrations": {
+            "lora_manager_enabled": bool(IntegrationConfig.LORA_MANAGER_ENABLED),
+            "has_civitai_api_key": bool(IntegrationConfig.CIVITAI_API_KEY),
+        },
+    }
+
+
+def _source_counts(sources):
+    configured = sources.get("configured") or []
+    effective = sources.get("effective") or []
+    return {
+        "configured": len(configured),
+        "effective": len(effective),
+        "watching": sum(1 for row in effective if row.get("watch") == "watching"),
+        "unavailable": sum(1 for row in configured if row.get("valid") is False),
+    }
+
+
+class DiagnosticsReportMixin:
+    """Mixin providing ``GET /prompt_manager/system/diagnostics``."""
+
+    def _register_diagnostics_report_routes(self, routes):
+        @routes.get("/prompt_manager/system/diagnostics")
+        async def diagnostics_report_route(request):
+            return await self.get_diagnostics_report(request)
+
+    async def get_diagnostics_report(self, request):
+        """The report, built off the loop under a lease (409 ``busy`` in a restore)."""
+        from ..config import IntegrationConfig
+
+        try:
+            async with self._jobs.lease(DIAGNOSTICS_LEASE):
+                report = await self._run_in_executor(self.build_report)
+        except JobBusy as busy:
+            return busy_response(busy)
+        except Exception:
+            self.logger.exception("The diagnostic report could not be built")
+            return web.json_response(
+                {
+                    "success": False,
+                    "code": None,
+                    "error": "The diagnostic report could not be built.",
+                },
+                status=500,
+            )
+        report = scrub(report, IntegrationConfig.CIVITAI_API_KEY)
+        return web.json_response(report, headers={"Cache-Control": "no-store"})
+
+    def build_report(self):
+        """The report's fields (blocking: reads the database and the log folder)."""
+        db = self.db
+        manager = self._get_logger_manager()
+        stats = dict(manager.get_log_stats())
+        stats["log_directory"] = report_path(stats.get("log_directory"))
+        info = dict(db.model.get_database_info() or {})
+        if "database_path" in info:
+            info["database_path"] = report_path(info["database_path"])
+        sources, _monitor = self._source_status()
+        return {
+            "success": True,
+            "generated_at": datetime.datetime.now().replace(microsecond=0).isoformat(),
+            "version": project_version(_ROOT),
+            "python": platform.python_version(),
+            "platform": {"system": platform.system(), "release": platform.release()},
+            "viewerjs": viewerjs_version(VIEWERJS_PROVENANCE),
+            "checks": _admin.legacy_checks(db.model.db_path, report_path),
+            "database_info": info,
+            "sources": _source_counts(sources),
+            "logging": {"config": dict(manager.get_config()), "stats": stats},
+            "health": health_summary(self._current_health_report()),
+            "config": public_config(),
+            "redacted": 0,
+        }

py/api/images.py

@@ -2272,8 +2272,7 @@ class ImageRoutesMixin:
                 )
 
             image_path = Path(image["image_path"])
-            lora_dirs = await self._run_in_executor(self._lora_image_dirs)
-            allowed_dirs = list(self._get_all_output_dirs()) + lora_dirs
+            allowed_dirs = list(self._get_all_output_dirs()) + self._lora_image_dirs()
 
             denied = _media_access_error(image_path, allowed_dirs)
             if denied is not None:
@@ -2321,10 +2320,7 @@ class ImageRoutesMixin:
         LoraManager's guarded LoRA roots, sidecar root and example-images
         root (``lora_manager_layout.media_roots``, #355) and PromptManager's
         CivitAI download cache. An unusable LoraManager layout gives the
-        default LoRA roots and the cache. The layout comes from the layout
-        cache (``lora_utils.cached_lora_manager_layout``, #362), which reloads
-        when LoraManager's settings files change; blocking, so the routes
-        call it in the executor.
+        default LoRA roots and the cache.
         """
         try:
             from ..config import IntegrationConfig
@@ -2332,18 +2328,25 @@ class ImageRoutesMixin:
             if not IntegrationConfig.LORA_MANAGER_ENABLED:
                 return []
             from ..lora_utils import (
-                cached_lora_manager_layout,
                 get_lora_image_cache_dir,
+                layout_for_install,
+                lora_manager_layout,
             )
 
-            configured = IntegrationConfig.LORA_MANAGER_PATH or ""
-            install, layout = cached_lora_manager_layout(configured)
-            if configured.strip() and install is None:
+            lora_root = IntegrationConfig.LORA_MANAGER_PATH or ""
+            if lora_root.strip():
                 # A configured path comes from config.json, which can be hand
                 # edited: only a LoraManager install under custom_nodes may
                 # widen the serving allow-list. An empty path keeps the
                 # auto-detection, which itself only scans custom_nodes.
-                return []
+                from ..lora_utils import resolve_lora_manager_path
+
+                lora_root = resolve_lora_manager_path(lora_root)
+                if not lora_root:
+                    return []
+                layout = layout_for_install(lora_root)
+            else:
+                layout = lora_manager_layout("")
 
             dirs = [Path(d) for d in layout.media_roots]
             # Named, not made: listing and serving never create the cache.

py/api/log_cleanup.py

@@ -0,0 +1,207 @@
+"""Reviewed rotated-log cleanup as a maintenance plan (#329 System S12).
+
+``POST /prompt_manager/maintenance/preview {operation: "rotated_logs",
+options: {}}`` lists the rotated copies of the active log that Rule R allows
+(``utils/logging_config.py``: ``prompt_manager.log.<n>`` regular files that
+are not links, directly in the log folder) and writes nothing. The plan keeps
+each copy's name, size, ``mtime_ns`` and inode under its token; the response
+is the maintenance ``Plan`` shape with basenames only, a ``none`` backup
+policy (no library record changes) and no ``disk`` key.
+
+``POST /prompt_manager/maintenance/apply {token}`` starts a ``maintenance``
+job titled "Rotated log copies" with admission ``read``: it may run beside a
+scan and is refused only beside another maintenance job or an ``exclusive``
+one. The job deletes exactly the reviewed copies under the logger's files
+lock and the file handler's lock, or nothing, failing ``plan-changed``, when
+any of them is missing, is no longer a rotated copy or differs in size,
+``mtime_ns`` or inode. A copy that cannot be deleted (open in another program
+on Windows) is an ``errors`` row; the others are deleted.
+"""
+
+import datetime
+import json
+
+from aiohttp import web
+
+try:
+    from ...utils.logging_config import RotatedChanged, rotated_sort_key
+except ImportError:  # loaded as the top-level ``py`` package (tests)
+    from utils.logging_config import RotatedChanged, rotated_sort_key
+
+from .jobs import MAINTENANCE_JOB_KIND
+from .maintenance_plans import ScopeTooLarge
+
+ROTATED_LOGS = "rotated_logs"
+LOG_CLEANUP_TITLE = "Rotated log copies"
+LOGS_CHANGED = (
+    "The log files changed since the review; nothing was deleted. Review again."
+)
+LOG_CLEANUP_FAILED = "The log cleanup failed."
+TOO_MANY_LOG_FILES = "Too many log files to review at once."
+LOG_CLEANUP_MEDIA = (
+    "Deletes only the listed log files. No image, video or library record changes."
+)
+_GROUPS = ("changed", "protected", "errors")
+
+
+def size_label(size):
+    """``1000 bytes``, ``2.0 KB``, ``10.0 MB`` or ``1.2 GB`` (1 KB = 1024 bytes)."""
+    size = max(0, int(size or 0))
+    if size < 1024:
+        return f"{size} {'byte' if size == 1 else 'bytes'}"
+    for unit, scale in (("KB", 2**10), ("MB", 2**20)):
+        if size < scale * 1024:
+            return f"{size / scale:.1f} {unit}"
+    return f"{size / 2**30:.1f} GB"
+
+
+def time_label(mtime_ns):
+    """The local modification time as ``YYYY-MM-DD HH:MM``."""
+    try:
+        moment = datetime.datetime.fromtimestamp(mtime_ns / 1e9)
+    except (OSError, OverflowError, ValueError):  # a time before 1970 on Windows
+        return "unknown time"
+    return moment.strftime("%Y-%m-%d %H:%M")
+
+
+def _count(count, singular, plural):
+    return f"{count} {singular if count == 1 else plural}"
+
+
+def _protected_reasons(rows):
+    """The plan's Protected reasons: what the cleanup keeps, zero counts left out."""
+    active = sum(1 for row in rows if row["kind"] == "active")
+    links = sum(1 for row in rows if row["kind"] == "link")
+    other = sum(1 for row in rows if row["kind"] == "other")
+    reasons = []
+    if active:
+        reasons.append("Kept: the active log")
+    if links:
+        reasons.append(f"Kept: {_count(links, 'link', 'links')}")
+    if other:
+        reasons.append(f"Kept: {_count(other, 'other file', 'other files')}")
+    return active + links + other, reasons
+
+
+def _scope_label(files):
+    if not files:
+        return "No rotated log copies"
+    total = sum(row["size"] for row in files)
+    return (
+        f"{_count(len(files), 'rotated log copy', 'rotated log copies')}"
+        f" · {size_label(total)}"
+    )
+
+
+class LogCleanupMixin:
+    """The ``rotated_logs`` maintenance plan and its job.
+
+    The logger manager is reached only through ``_get_logger_manager``
+    (``logging_routes.py``), so the tests can point it at a temporary folder.
+    """
+
+    async def preview_log_cleanup(self):
+        """The ``rotated_logs`` plan (runs inside the preview's lease)."""
+        from . import maintenance
+
+        manager = self._get_logger_manager()
+        try:
+            rows = await self._run_in_executor(manager.scan_log_folder)
+        except Exception:
+            self.logger.exception("Rotated log review failed")
+            return maintenance._failure("Failed to review maintenance", 500)
+        rotated = sorted(
+            (row for row in rows if row["kind"] == "rotated"),
+            key=rotated_sort_key,
+        )
+        files = [
+            {
+                "name": row["name"],
+                "size": row["size"],
+                "mtime_ns": row["mtime_ns"],
+                "ino": row["ino"],
+            }
+            for row in rotated
+        ]
+        protected, reasons = _protected_reasons(rows)
+        store = self._maintenance_plans
+        try:
+            token = store.issue(
+                ROTATED_LOGS,
+                json.dumps(files, sort_keys=True),
+                {"files": files, "protected": protected},
+                (),
+            )
+        except ScopeTooLarge:
+            return maintenance._failure(TOO_MANY_LOG_FILES, 413, "too-large")
+        plan = store.peek(token, ROTATED_LOGS)
+        return web.json_response(
+            {
+                "success": True,
+                "token": token,
+                "operation": ROTATED_LOGS,
+                "expires_at": maintenance._iso(plan.expires_at),
+                "scope": {"count": len(files), "label": _scope_label(files)},
+                "protected": {"count": protected, "reasons": reasons},
+                "backup": {
+                    "policy": "none",
+                    "destination_label": None,
+                    "expected_bytes": None,
+                },
+                "media": {"deletes_files": True, "label": LOG_CLEANUP_MEDIA},
+                "details": [
+                    {
+                        "label": row["name"],
+                        "detail": (
+                            f"{size_label(row['size'])} · {time_label(row['mtime_ns'])}"
+                        ),
+                    }
+                    for row in files
+                ],
+            },
+            headers={"Cache-Control": "no-store"},
+        )
+
+    def _start_log_cleanup(self, plan):
+        """Register the cleanup job (on the loop); raises JobBusy when refused."""
+        return self._jobs.start(
+            MAINTENANCE_JOB_KIND,
+            LOG_CLEANUP_TITLE,
+            admission="read",
+            cancellable=False,
+            scope={"operation": ROTATED_LOGS},
+            groups=_GROUPS,
+            runner=lambda job: self._run_log_cleanup(job, plan),
+        )
+
+    async def _run_log_cleanup(self, job, plan):
+        from . import maintenance
+
+        files = plan.payload["files"]
+        sizes = {row["name"]: row["size"] for row in files}
+        job.update(total=len(files), status_line="Deleting rotated log copies")
+        manager = self._get_logger_manager()
+        try:
+            result = await self._run_in_executor(manager.delete_rotated, files)
+        except RotatedChanged:
+            return maintenance._fail_job(job, "plan-changed", LOGS_CHANGED)
+        except Exception:
+            self.logger.exception("Rotated log cleanup failed")
+            return maintenance._fail_job(job, None, LOG_CLEANUP_FAILED)
+        for name in result["removed"]:
+            detail = f"Deleted · {size_label(sizes.get(name, 0))}"
+            job.record(
+                "changed", {"path": name, "detail": detail}, f"- {name} · {detail}"
+            )
+        for name, error in result["failed"]:
+            detail = self._public_error(error)
+            job.record(
+                "errors", {"path": name, "detail": detail}, f"! {name} · {detail}"
+            )
+        job.record_counted("protected", plan.payload["protected"])
+        removed, failed = len(result["removed"]), len(result["failed"])
+        line = f"Deleted {_count(removed, 'rotated log copy', 'rotated log copies')}"
+        if failed:
+            line += f", {failed} could not be deleted"
+        job.log_line(line)
+        job.update(processed=len(files), progress=100, status_line=line)

py/api/logging_routes.py

@@ -1,6 +1,13 @@
-"""Logging API routes for PromptManager."""
+"""Logging API routes for PromptManager.
+
+Every route that names a log file goes through Rule R (#329,
+``utils/logging_config.py``): a download is served only for a regular,
+non-link file listed directly in the log folder, and Clear deletes only
+rotated copies. File work runs off the event loop.
+"""
 
 import os
+import stat
 
 from aiohttp import web
 
@@ -85,6 +92,41 @@ def _validate_log_config(data):
     return clean, None
 
 
+def _read_log_bytes(path):
+    """The whole file at ``path``, opened without following a link (blocking).
+
+    The file is checked again through the open handle: a regular file with
+    one name and, when ``path`` carries it, the inode it was listed with. A
+    name swapped for a link or another file after it was listed, or a hard
+    link to a file elsewhere, is refused as missing. The handle is closed
+    before this returns.
+    """
+    flags = os.O_RDONLY | getattr(os, "O_BINARY", 0) | getattr(os, "O_NOFOLLOW", 0)
+    # A FIFO swapped in after the listing must not block the executor thread.
+    flags |= getattr(os, "O_NONBLOCK", 0)
+    try:
+        fd = os.open(str(path), flags)
+    except OSError as error:
+        if getattr(os, "O_NOFOLLOW", 0) and os.path.islink(str(path)):
+            raise FileNotFoundError("Log file not found") from error
+        raise
+    listed_ino = getattr(path, "listed_ino", 0)
+    with os.fdopen(fd, "rb") as handle:
+        opened = os.fstat(handle.fileno())
+        if (
+            not stat.S_ISREG(opened.st_mode)
+            or opened.st_nlink > 1  # a hard link to a file elsewhere
+            or (listed_ino and opened.st_ino and opened.st_ino != listed_ino)
+        ):
+            raise FileNotFoundError("Log file not found")
+        return handle.read()
+
+
+def _download(logger_manager, filename):
+    """``bytes`` of a downloadable log file; FileNotFoundError otherwise."""
+    return _read_log_bytes(logger_manager.log_file_entry(filename))
+
+
 class LoggingRoutesMixin:
     """Mixin providing logging-related API endpoints."""
 
@@ -195,16 +237,16 @@ class LoggingRoutesMixin:
                     {"success": False, "error": "Invalid filename"}, status=400
                 )
 
-            log_file_path = logger_manager.log_dir / filename
-
-            if not log_file_path.exists():
+            try:
+                file_content = await self._run_in_executor(
+                    _download, logger_manager, filename
+                )
+            except FileNotFoundError:
+                # Missing, a link, a folder or not listed: all read as absent.
                 return web.json_response(
                     {"success": False, "error": "Log file not found"}, status=404
                 )
 
-            with open(log_file_path, "rb") as f:
-                file_content = f.read()
-
             return web.Response(
                 body=file_content,
                 content_type="text/plain",
@@ -224,7 +266,7 @@ class LoggingRoutesMixin:
         """Truncate all log files."""
         try:
             logger_manager = self._get_logger_manager()
-            results = logger_manager.truncate_logs()
+            results = await self._run_in_executor(logger_manager.truncate_logs)
 
             return web.json_response(
                 {
@@ -266,7 +308,7 @@ class LoggingRoutesMixin:
                 return web.json_response({"success": False, "error": error}, status=400)
 
             logger_manager = self._get_logger_manager()
-            logger_manager.update_config(clean)
+            await self._run_in_executor(logger_manager.update_config, clean)
 
             return web.json_response(
                 {

py/api/lora_import_routes.py

@@ -194,13 +194,8 @@ class LoraImportRoutesMixin:
             return False
 
     async def replace_lora_imports(self, db, staged, *, expected_ids, allow_empty):
-        """Swap the collection for ``staged.records`` in one transaction.
-
-        Then the image routes' layout cache is dropped (#362): the import may
-        have linked files under a folder LoraManager created after the cache
-        stored its layout (its first example-image download creates one).
-        """
-        result = await self._run_in_executor(
+        """Swap the collection for ``staged.records`` in one transaction."""
+        return await self._run_in_executor(
             lambda: db.replace_category_prompts(
                 CATEGORY,
                 staged.records,
@@ -208,8 +203,6 @@ class LoraImportRoutesMixin:
                 allow_empty=allow_empty,
             )
         )
-        lora_utils.clear_layout_cache()
-        return result
 
     # ── preview ───────────────────────────────────────────────────────
 

py/api/maintenance.py

@@ -2,7 +2,9 @@
 
 ``POST /prompt_manager/maintenance/preview {operation, options:{}}`` reviews
 one operation (``duplicate_prompts``, ``optimize``, ``unused_components`` or
-``missing_links``; ``database/maintenance.py``) off the event loop and
+``missing_links``; ``database/maintenance.py``; ``duplicate_files``,
+``py/api/duplicate_files.py``; ``rotated_logs``, the rotated log copies,
+``py/api/log_cleanup.py``, which has its own apply job) off the event loop and
 answers a ``Plan``: ``{success, token, operation, expires_at, scope:{count,
 label}, protected:{count, reasons}, backup:{policy, destination_label},
 media:{deletes_files, label}, details:[≤50 {label, detail}]}``. The preview
@@ -54,6 +56,7 @@ from aiohttp import web
 from .admin import _prune_safety_backups, _scan_location, _scan_roots
 from .job_routes import busy_response
 from .jobs import MAINTENANCE_JOB_KIND, JobBusy
+from .log_cleanup import ROTATED_LOGS
 from .maintenance_plans import PlanExpired, PlanUsed, ScopeTooLarge
 
 try:
@@ -83,7 +86,8 @@ MAINTENANCE_PREVIEW_LEASE = "Maintenance preview"
 DUPLICATE_PREVIEW_BODY_MAX = 1024 * 1024
 DUPLICATE_FILES = "duplicate_files"
 WORKFLOW_STORAGE = "workflow_storage"  # applied by py/api/workflow_storage.py
-PREVIEW_OPERATIONS = MAINTENANCE_OPERATIONS + (DUPLICATE_FILES,)
+# rotated_logs (#329, py/api/log_cleanup.py) reviews log files, not records.
+PREVIEW_OPERATIONS = MAINTENANCE_OPERATIONS + (DUPLICATE_FILES, ROTATED_LOGS)
 # Room left over beyond what a step writes, so a disk is never filled to 0.
 DISK_MARGIN_BYTES = 16 * 2**20
 # The concept's Maintenance row titles, which name each operation's job.
@@ -391,6 +395,8 @@ class MaintenanceRoutesMixin:
             return await self.preview_duplicate_files(options)
         if not isinstance(options, dict) or options:
             return _invalid(f"{operation} takes no options")
+        if operation == ROTATED_LOGS:
+            return await self.preview_log_cleanup()
         db = self.db
         identity = self._database_identity(db)
         try:
@@ -492,6 +498,17 @@ class MaintenanceRoutesMixin:
         if refused is not None:
             return refused
         store = self._maintenance_plans
+        if plan.operation == ROTATED_LOGS:
+            # Log files only: no database identity, backup or disk check.
+            try:
+                job = self._start_log_cleanup(plan)
+            except JobBusy as busy:
+                return busy_response(busy)  # the token stays usable
+            # Same loop turn as the peek and the start: the runner has not run.
+            store.consume(token, plan.operation, job_id=job.id)
+            return web.json_response(
+                {"success": True, "job": job.summary()}, status=202
+            )
         db = self.db
         if plan.db_identity != self._database_identity(db):
             store.discard(token)

py/lora_import.py

@@ -12,10 +12,9 @@ last writes to the database:
    the sidecars (``discover``). In ``alongside`` mode a sidecar beside its
    model is kept when ``<base>.safetensors`` exists; in ``centralized`` mode
    every sidecar under the sidecar root is kept when its ``file_path`` names
-   a model inside a LoRA root. ``fingerprint`` covers every scanned sidecar,
-   kept or skipped (path, size, modification time, kept or not, #362) and the
-   layout; ``preview`` adds the counts a review shows. Nothing is written
-   and nothing is requested.
+   a model inside a LoRA root. ``fingerprint`` covers the kept sidecars (path,
+   size and modification time) and the layout; ``preview`` adds the counts a
+   review shows. Nothing is written and nothing is requested.
 2. **stage**: read each sidecar and build its record. Its LoraManager preview
    and example images are linked where they are (``local_media``); only a
    LoRA without any local image downloads its CivitAI example images into the
@@ -70,10 +69,9 @@ def discover(lora_manager_path: Optional[str]) -> Discovery:
 
 
 def fingerprint(discovery: Discovery) -> str:
-    """What a review saw: the layout, then every scanned sidecar, kept or
-    skipped, with its path, size, modification time and whether it was kept
-    (#362). A LoraManager save that changes unrelated settings leaves it as
-    it was."""
+    """What a review saw: the layout, then each kept sidecar's path, size and
+    modification time (a LoraManager save that changes unrelated settings
+    leaves it as it was)."""
     layout = discovery.layout
     digest = hashlib.sha256()
     for value in (
@@ -89,14 +87,13 @@ def fingerprint(discovery: Discovery) -> str:
         digest.update(f"{value}\n".encode("utf-8", "surrogatepass"))
     for name in discovery.unreadable:
         digest.update(f"unreadable\0{name}\n".encode("utf-8", "surrogatepass"))
-    kept = {str(sidecar.metadata_path) for sidecar in discovery.sidecars}
-    for path in discovery.scanned_files:
-        flag = 1 if str(path) in kept else 0
+    for sidecar in discovery.sidecars:
+        path = sidecar.metadata_path
         try:
             stat = path.stat()
-            entry = f"{path}\0{stat.st_size}\0{stat.st_mtime_ns}\0{flag}\n"
+            entry = f"{path}\0{stat.st_size}\0{stat.st_mtime_ns}\n"
         except OSError:
-            entry = f"{path}\0missing\0{flag}\n"
+            entry = f"{path}\0missing\n"
         digest.update(entry.encode("utf-8", "surrogatepass"))
     return "sha256:" + digest.hexdigest()
 

py/lora_layout_cache.py

@@ -1,217 +0,0 @@
-"""A per-process cache of LoraManager's layout for the image hot paths (#362).
-
-Every Library list, prompt detail, inspect, reveal and serve request needs the
-LoRA allow-list (``ImagesMixin._lora_image_dirs``). Building it means finding
-the LoraManager install, reading its plugin settings file and its settings
-file, and resolving every folder. ``LayoutCache`` keeps the last result and
-reuses it while all of these hold:
-
-- the key (the configured install path, stripped) and the home folder the
-  folder guard compares against are the same;
-- the entry is younger than ``MAX_AGE_SECONDS`` (monotonic clock);
-- the plugin settings file and the settings file have the same signature
-  (``st_mtime_ns``, ``st_size``, or ``MISSING``) as when it was stored;
-- the install and every stored media root still resolve to themselves (a
-  folder swapped for a symlink reloads, so the folder guard runs again).
-
-A result is stored only when no watched file is "racy" (modified within
-``RACY_SECONDS`` of the load, git's rule for racily clean index entries),
-every watched file could be checked, the settings file was readable, and no
-``clear()`` ran during the load. What each rule covers: signatures catch a
-settings change on the next request; the racy rule catches a write during the
-load, a half-written file and coarse timestamps; the maximum age bounds the
-inputs no signature sees (``extra_model_paths.yaml``, environment, temporary
-failures); ``clear()`` makes a finished import's new folders servable at once.
-
-Hits and loads take no lock: a slow ``os.stat`` on a network share holds up
-only its own request, and two threads that miss together both load (both
-loads only read). A short lock guards the compare-and-store and the logging
-state; no file I/O runs under it. Standard library only, so it is tested
-without ComfyUI or LoraManager.
-"""
-
-import logging
-import os
-import threading
-import time
-from dataclasses import dataclass
-from pathlib import Path
-from typing import Callable, Optional, Tuple
-
-from .lora_manager_layout import UNREADABLE, LoraManagerLayout, _canonical
-
-# A watched file modified this recently is not trusted: LoraManager writes its
-# settings in place, and FAT/exFAT store modification times in 2 s steps.
-RACY_SECONDS = 3
-# Bounds what no signature sees (unwatched inputs, temporary failures).
-MAX_AGE_SECONDS = 30
-
-# The signature of a watched file that does not exist (a valid state).
-MISSING = object()
-# ERROR_BAD_NETPATH and ERROR_BAD_NET_NAME, which Windows raises as
-# FileNotFoundError.
-_OFFLINE_SHARE_ERRORS = frozenset({53, 67})
-
-logger = logging.getLogger("prompt_manager.lora_layout_cache")
-
-
-@dataclass(frozen=True)
-class _Entry:
-    key: str
-    home: str
-    install: Optional[str]
-    layout: LoraManagerLayout
-    signatures: Tuple[Tuple[str, object], ...]
-    stored_at: float
-
-
-class _CannotValidate(Exception):
-    """A watched file could not be checked (not merely missing)."""
-
-
-class LayoutCache:
-    """At most one ``(install, layout)`` for the active LoraManager config."""
-
-    def __init__(
-        self,
-        resolve: Callable[[str], Optional[str]],
-        load: Callable[[Optional[str]], LoraManagerLayout],
-        *,
-        wall_ns: Callable[[], int] = time.time_ns,
-        monotonic: Callable[[], float] = time.monotonic,
-        stat=os.stat,
-        pathmod=os.path,
-        home: Callable[[], str] = lambda: str(Path.home()),
-    ):
-        self._resolve = resolve
-        self._load = load
-        self._wall_ns = wall_ns
-        self._monotonic = monotonic
-        self._stat = stat
-        self._pathmod = pathmod
-        self._home = home
-        self._entry: Optional[_Entry] = None
-        self._generation = 0
-        self._lock = threading.Lock()
-        # What was last logged; survives clear() so a reload of the same
-        # layout stays quiet.
-        self._logged_layout: Optional[LoraManagerLayout] = None
-        self._logged_refused: Tuple[str, ...] = ()
-
-    def get(self, configured: str) -> Tuple[Optional[str], LoraManagerLayout]:
-        """``(install, layout)`` for the configured path ("" auto-detects)."""
-        key = (configured or "").strip()
-        home = self._home()
-        entry = self._entry
-        if entry is not None and self._valid(entry, key, home):
-            return entry.install, entry.layout
-        with self._lock:
-            if entry is not None and self._entry is entry:
-                self._entry = None
-            generation = self._generation
-        install = self._resolve(key)
-        layout = self._load(install)
-        self._log_refused(layout)
-        try:
-            signatures = tuple(
-                (path, self._signature(path)) for path in self._watched(install, layout)
-            )
-        except _CannotValidate:
-            return install, layout
-        if layout.error == UNREADABLE or self._racy(signatures, self._wall_ns()):
-            return install, layout
-        stored = _Entry(key, home, install, layout, signatures, self._monotonic())
-        changed = False
-        with self._lock:
-            if self._generation == generation:
-                self._entry = stored
-                changed = layout != self._logged_layout
-                if changed:
-                    self._logged_layout = layout
-        if changed:
-            logger.info(
-                f"LoraManager layout changed: {layout.source or 'no'} settings, "
-                f"{layout.mode} storage, {len(layout.media_roots)} media folders"
-            )
-        return install, layout
-
-    def clear(self) -> None:
-        """Drop the entry; a load that started before this stores nothing."""
-        with self._lock:
-            self._entry = None
-            self._generation += 1
-
-    def _log_refused(self, layout: LoraManagerLayout) -> None:
-        """One warning per default LoRA folder newly refused by the guard."""
-        refused = tuple(layout.refused_defaults)
-        with self._lock:
-            if refused == self._logged_refused:
-                return
-            new = [folder for folder in refused if folder not in self._logged_refused]
-            self._logged_refused = refused
-        for folder in new:
-            logger.warning(
-                f"ComfyUI LoRA folder '{folder}' is neither scanned nor served: "
-                "it is a filesystem root, the home folder or a folder that "
-                "contains it"
-            )
-
-    # ── validation ──────────────────────────────────────────────────
-
-    def _valid(self, entry: _Entry, key: str, home: str) -> bool:
-        if entry.key != key or entry.home != home:
-            return False
-        if self._monotonic() - entry.stored_at >= MAX_AGE_SECONDS:
-            return False
-        try:
-            for path, signature in entry.signatures:
-                if self._signature(path) != signature:
-                    return False
-        except _CannotValidate:
-            return False
-        folders = entry.layout.media_roots
-        if entry.install is not None:
-            folders = (entry.install,) + tuple(folders)
-        for folder in folders:
-            try:
-                if _canonical(folder, self._pathmod) != folder:
-                    return False
-            except (OSError, ValueError):
-                return False
-        return True
-
-    def _watched(self, install, layout) -> Tuple[str, ...]:
-        """The plugin settings file, then the settings file, each once."""
-        if install is None:
-            return ()
-        paths = [self._pathmod.join(install, "settings.json")]
-        if layout.settings_path is not None:
-            paths.append(str(layout.settings_path))
-        watched, seen = [], set()
-        for path in paths:
-            identity = self._pathmod.normcase(self._pathmod.abspath(path))
-            if identity not in seen:
-                seen.add(identity)
-                watched.append(path)
-        return tuple(watched)
-
-    def _signature(self, path: str):
-        try:
-            result = self._stat(path)
-        except (FileNotFoundError, NotADirectoryError) as error:
-            # Windows reports an offline share as "not found": that is a
-            # temporary failure, not a missing file.
-            if getattr(error, "winerror", None) in _OFFLINE_SHARE_ERRORS:
-                raise _CannotValidate(path) from error
-            return MISSING
-        except (OSError, ValueError) as error:
-            raise _CannotValidate(path) from error
-        return (result.st_mtime_ns, result.st_size)
-
-    @staticmethod
-    def _racy(signatures, now_ns: int) -> bool:
-        threshold = now_ns - RACY_SECONDS * 10**9
-        return any(
-            signature is not MISSING and signature[0] > threshold
-            for _path, signature in signatures
-        )

py/lora_manager_layout.py

@@ -12,10 +12,8 @@ server share one definition:
   plugin file, or the user config folder, in LoraManager's order), keeps only
   the seven keys the import needs, and resolves the LoRA roots, the sidecar
   root (centralized storage), the example-images root and the active
-  LoraManager library. Every folder from the settings, and each of
-  ComfyUI's default LoRA folders, passes the folder guard: it must exist and
-  must not be a filesystem root, the home folder or a folder that contains
-  it. Default folders the guard refuses are listed in ``refused_defaults``.
+  LoraManager library. Every folder from the settings passes the folder
+  guard (exists, not a filesystem root, not the home folder itself).
 - ``discover_sidecars`` finds every sidecar whose model file exists inside a
   LoRA root: beside the models in ``alongside`` mode, under the sidecar root
   (matched to its model by ``file_path``) in ``centralized`` mode.
@@ -27,7 +25,6 @@ it is testable outside ComfyUI.
 """
 
 import json
-import logging
 import os
 import re
 from dataclasses import dataclass, replace
@@ -101,12 +98,7 @@ NOT_OBJECT = "the settings file is not a JSON object"
 TOO_LARGE = "the settings file is larger than 5 MB"
 NO_PLATFORMDIRS = "platformdirs is not available"
 NO_SIDECAR_ROOT = "the sidecar folder does not exist"
-BAD_SIDECAR_ROOT = (
-    "the sidecar folder is a filesystem root, the home folder or a folder "
-    "that contains it"
-)
-
-logger = logging.getLogger("prompt_manager.lora_manager_layout")
+BAD_SIDECAR_ROOT = "the sidecar folder is a filesystem root or the home folder"
 
 
 @dataclass(frozen=True)
@@ -122,9 +114,6 @@ class LoraManagerLayout:
     example_images_root: Optional[str]
     library: str
     media_roots: Tuple[str, ...]
-    # ComfyUI's default LoRA folders that exist but that the folder guard
-    # refused (canonical): the layout cache logs them (#362).
-    refused_defaults: Tuple[str, ...] = ()
 
 
 @dataclass(frozen=True)
@@ -137,20 +126,13 @@ class Sidecar:
 
 @dataclass(frozen=True)
 class Discovery:
-    """The sidecars kept, every metadata file seen, and those unread."""
+    """The sidecars kept, how many metadata files were seen, and those unread."""
 
     layout: LoraManagerLayout
     sidecars: List[Sidecar]
-    # Every ``*.metadata.json`` seen, kept or skipped: resolved, once each,
-    # sorted by ``str(path)``. The import's fingerprint covers all of them.
-    scanned_files: Tuple[Path, ...]
+    scanned: int
     unreadable: List[str]
 
-    @property
-    def scanned(self) -> int:
-        """How many metadata files were seen."""
-        return len(self.scanned_files)
-
 
 class _SettingsError(Exception):
     """The settings file cannot be used, for the reason it carries."""
@@ -333,23 +315,6 @@ def _guarded(paths, home: str) -> Tuple[str, ...]:
     return tuple(kept)
 
 
-def _refused(paths, home: str) -> Tuple[str, ...]:
-    """The folders in ``paths`` that exist but that the guard refuses."""
-    refused = []
-    for path in paths:
-        absolute = _absolute(path)
-        if not absolute or not os.path.isdir(absolute):
-            continue
-        if _guard_folder(absolute, home) is None:
-            try:
-                canonical = _canonical(absolute)
-            except (OSError, ValueError):
-                continue
-            if canonical not in refused:
-                refused.append(canonical)
-    return tuple(refused)
-
-
 def _example_images_root(
     settings: Mapping, library: str, libraries: int, home: str
 ) -> Optional[str]:
@@ -386,9 +351,7 @@ def load_layout(
     default LoRA roots, so serving and trigger words keep working.
     """
     home = str(Path.home())
-    default_list = list(default_roots())
-    defaults = _guarded(default_list, home)
-    refused_defaults = _refused(default_list, home)
+    defaults = _guarded(default_roots(), home)
     fallback = LoraManagerLayout(
         settings_path=None,
         source=None,
@@ -399,7 +362,6 @@ def load_layout(
         example_images_root=None,
         library="default",
         media_roots=defaults,
-        refused_defaults=refused_defaults,
     )
     if not lora_manager_path:
         return fallback
@@ -415,14 +377,10 @@ def load_layout(
     except _SettingsError as error:
         return replace(fallback, error=str(error))
     settings = _settings_subset(raw or {})
-    return _layout_from_settings(
-        settings, settings_path, source, defaults, home, refused_defaults
-    )
+    return _layout_from_settings(settings, settings_path, source, defaults, home)
 
 
-def _layout_from_settings(
-    settings, settings_path, source, defaults, home, refused_defaults=()
-):
+def _layout_from_settings(settings, settings_path, source, defaults, home):
     library, entry, libraries = _active_library(settings)
     lora_roots = _media_roots(
         defaults,
@@ -442,7 +400,6 @@ def _layout_from_settings(
         example_images_root=example_root,
         library=library,
         media_roots=_media_roots(lora_roots, [example_root]),
-        refused_defaults=refused_defaults,
     )
     if mode != MODE_CENTRALIZED:
         return layout
@@ -525,13 +482,13 @@ def discover_sidecars(layout: LoraManagerLayout) -> Discovery:
     cannot see every folder must not replace the collection with part of it.
     """
     if layout.error is not None:
-        return Discovery(layout, [], (), [])
+        return Discovery(layout, [], 0, [])
     if layout.mode == MODE_CENTRALIZED:
         folders = [layout.sidecar_root]
     else:
         folders = list(layout.lora_roots)
     seen = set()
-    scanned: List[Path] = []
+    scanned = 0
     unreadable: List[str] = []
     sidecars = []
     for folder in folders:
@@ -541,13 +498,12 @@ def discover_sidecars(layout: LoraManagerLayout) -> Discovery:
             if key in seen:
                 continue
             seen.add(key)
-            scanned.append(resolved)
+            scanned += 1
             model = _model_for(resolved, layout, unreadable)
             if model is not None:
                 sidecars.append(Sidecar(resolved, Path(os.path.realpath(model))))
     sidecars.sort(key=lambda sidecar: str(sidecar.metadata_path))
-    scanned.sort(key=str)
-    return Discovery(layout, sidecars, tuple(scanned), unreadable)
+    return Discovery(layout, sidecars, scanned, unreadable)
 
 
 # ── images ──────────────────────────────────────────────────────────
@@ -570,15 +526,8 @@ def _example_files(folder: str) -> List[str]:
 def _preview_candidates(sidecar: Sidecar, metadata: Dict) -> Iterator[str]:
     """The paths a preview is looked for at, in LoraManager's order."""
     url = metadata.get("preview_url")
-    if isinstance(url, str) and url.strip():
-        if os.path.isabs(url.strip()):
-            yield url.strip()
-        else:
-            # LoraManager writes absolute paths; it would resolve a relative
-            # one against its working directory, not the sidecar folder.
-            logger.debug(
-                "Relative preview_url skipped in %s", sidecar.metadata_path.name
-            )
+    if isinstance(url, str) and url.strip() and os.path.isabs(url.strip()):
+        yield url.strip()
     base = sidecar.metadata_path.name[: -len(METADATA_SUFFIX)]
     for suffix in PREVIEW_EXTENSIONS:
         yield str(sidecar.metadata_path.parent / f"{base}{suffix}")

py/lora_utils.py

@@ -30,7 +30,6 @@ except ImportError:
 
 from dataclasses import replace
 
-from .lora_layout_cache import LayoutCache
 from .lora_manager_layout import (
     MODE_ALONGSIDE,
     METADATA_SUFFIX,
@@ -296,35 +295,10 @@ def layout_for_install(install: Optional[str]) -> LoraManagerLayout:
 
 
 def lora_manager_layout(lora_manager_path: Optional[str]) -> LoraManagerLayout:
-    """Where LoraManager keeps its files (``py/lora_manager_layout.py``).
-
-    A fresh read: the import review, Start, the legacy scan and the trigger
-    word cache use it. The image hot paths use ``cached_lora_manager_layout``.
-    """
+    """Where LoraManager keeps its files (``py/lora_manager_layout.py``)."""
     return layout_for_install(lora_manager_install(lora_manager_path))
 
 
-# The image hot paths' layout cache (#362). The lambdas look the functions up
-# when called, so a test that patches them here is seen by the cache.
-_layout_cache = LayoutCache(
-    lambda path: lora_manager_install(path),
-    lambda install: layout_for_install(install),
-)
-
-
-def cached_lora_manager_layout(
-    configured: Optional[str],
-) -> Tuple[Optional[str], LoraManagerLayout]:
-    """``(install, layout)`` for the configured path, from the layout cache
-    while LoraManager's settings files are unchanged (``py/lora_layout_cache.py``)."""
-    return _layout_cache.get(configured or "")
-
-
-def clear_layout_cache() -> None:
-    """Drop the cached layout (after an import linked files in new folders)."""
-    _layout_cache.clear()
-
-
 def _get_extra_lora_paths(comfyui_root: Path) -> List[Path]:
     """Parse extra_model_paths.yaml for additional LoRA directories."""
     results = []
@@ -368,7 +342,7 @@ def read_lora_metadata(metadata_path: Path) -> Optional[Dict]:
     try:
         with open(metadata_path, "r", encoding="utf-8") as f:
             return json.load(f)
-    except (ValueError, OSError) as e:  # JSONDecodeError, UnicodeDecodeError
+    except (json.JSONDecodeError, OSError) as e:
         logger.debug(f"Failed to read {metadata_path}: {e}")
         return None
 
@@ -883,13 +857,11 @@ class TriggerWordCache:
         """
         try:
             new_cache = self._build(lora_manager_path)
-        except OSError as error:
+        except Exception as error:
             logger.warning(f"Trigger word cache not refreshed: {error}")
-            return self._failed()
-        except Exception:
-            # Not a disk error: keep the traceback (#362).
-            logger.exception("Trigger word cache not refreshed")
-            return self._failed()
+            with self._lock:
+                self._failed_at = time.monotonic()
+                return len(self._cache)
 
         with self._lock:
             self._cache = new_cache
@@ -901,12 +873,6 @@ class TriggerWordCache:
         )
         return len(new_cache)
 
-    def _failed(self) -> int:
-        """Keep the previous map and note when this load failed."""
-        with self._lock:
-            self._failed_at = time.monotonic()
-            return len(self._cache)
-
     @staticmethod
     def _build(lora_manager_path: str) -> Dict[str, List[str]]:
         layout = lora_manager_layout(lora_manager_path)

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

@@ -429,9 +429,7 @@ test('Rebuild all is explicit and the plan line says every preview is regenerate
     await expect(page).toHaveURL(/#\/activity\/[0-9a-f]{32}$/);
     expect(seen.starts).toEqual([{ mode: 'rebuild' }]);
     const id = page.url().split('/').pop();
-    // Rebuild regenerates all 24 previews with real image work: on a busy CI
-    // runner that took over 5 s (12 of 24 done at the default poll budget).
-    await expect.poll(async () => (await serverJson(request, `/prompt_manager/jobs/${id}`)).job.state, { timeout: 30000 }).toBe('finished');
+    await expect.poll(async () => (await serverJson(request, `/prompt_manager/jobs/${id}`)).job.state).toBe('finished');
 });
 
 test('a plan that cannot be read says so; Start still starts', async ({ page }) => {

tests/comfyui/fixtures/ledger_nodes.py

@@ -292,8 +292,7 @@ async def image_monitor(request: web.Request) -> web.Response:
         config.MONITORING_ENABLED = True
         monitor.start_monitoring([folder_paths.get_output_directory()])
     else:
-        # Joins the image worker and the observer (one shared 5 s deadline),
-        # so it runs on a worker thread, not on the event loop.
+        # Joins the image worker and the observer (up to 5 s each): off the loop.
         await asyncio.to_thread(monitor.stop_monitoring)
         config.MONITORING_ENABLED = False
     return web.json_response({"monitoring": monitor.is_monitoring})

tests/test_api_diagnostics_report.py

@@ -0,0 +1,409 @@
+"""The shareable diagnostic report (#329, py/api/diagnostics_report.py).
+
+``GET /prompt_manager/system/diagnostics`` builds the report on the server
+from an explicit list of fields: the project version, the four legacy checks,
+public paths only and no secret. A scrub replaces anything that still looks
+like an absolute path or holds the CivitAI key and counts it in ``redacted``;
+these tests require ``redacted == 0`` so a builder mistake fails instead of
+being hidden.
+"""
+
+import json
+import ntpath
+import os
+import posixpath
+import re
+import sqlite3
+import sys
+import unittest
+from unittest import mock
+from unittest.mock import MagicMock
+
+_mock_server = MagicMock()
+_mock_server.PromptServer.instance.routes = MagicMock()
+sys.modules.setdefault("server", _mock_server)
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+from py.api import admin, diagnostics_report  # noqa: E402
+from py.api.diagnostics_report import OUTSIDE, report_path, scrub  # noqa: E402
+from py.config import (
+    GalleryConfig,
+    IntegrationConfig,
+    PromptManagerConfig,
+)  # noqa: E402
+
+try:
+    from tests.test_api_log_cleanup import LogCleanupTestCase  # noqa: E402
+except ImportError:  # discovered with tests/ as the top-level directory
+    from test_api_log_cleanup import LogCleanupTestCase  # noqa: E402
+
+REPORT = "/prompt_manager/system/diagnostics"
+KEY = "civitai-secret-0123456789abcdef"
+SEEDED_PATHS = (
+    "/home/someone/private/LoraManager",
+    "/srv/someone/renders",
+    "C:\\Users\\someone\\Documents\\renders",
+    "D:/Users/someone/prompts.db",
+    "\\\\nas\\share\\someone\\renders",
+)
+_WINDOWS_ABSOLUTE = re.compile(r"^[A-Za-z]:[\\/]")
+TOP_KEYS = {
+    "success",
+    "generated_at",
+    "version",
+    "python",
+    "platform",
+    "viewerjs",
+    "checks",
+    "database_info",
+    "sources",
+    "logging",
+    "health",
+    "config",
+    "redacted",
+}
+CONFIG = {
+    "database": {"enable_duplicate_detection", "enable_auto_save"},
+    "web_ui": {
+        "result_timeout",
+        "show_test_button",
+        "webui_display_mode",
+        "infinite_scroll",
+    },
+    "performance": {
+        "max_search_results",
+        "enable_fuzzy_search",
+        "auto_backup_interval",
+        "worker_threads",
+    },
+    "gallery": {
+        "processing_delay",
+        "prompt_timeout",
+        "cleanup_interval",
+        "auto_cleanup_missing_files",
+        "max_image_age_days",
+        "max_concurrent_processing",
+        "metadata_extraction_timeout",
+    },
+    "integrations": {"lora_manager_enabled", "has_civitai_api_key"},
+}
+
+
+def _strings(value):
+    if isinstance(value, str):
+        yield value
+    elif isinstance(value, dict):
+        for key, item in value.items():
+            yield str(key)
+            yield from _strings(item)
+    elif isinstance(value, (list, tuple)):
+        for item in value:
+            yield from _strings(item)
+
+
+class DiagnosticsReportTestCase(LogCleanupTestCase):
+    """The admin app with its logger manager on a temporary log folder."""
+
+    async def get_application(self):
+        app = await super().get_application()
+        saved = IntegrationConfig.get_config()
+        self.addCleanup(IntegrationConfig.update_config, saved)
+        return app
+
+    async def report(self, status=200):
+        response = await self.client.request("GET", REPORT)
+        text = await response.text()
+        self.assertEqual(response.status, status, text)
+        return json.loads(text), text
+
+    def seed_secrets(self):
+        IntegrationConfig.update_config(
+            {
+                "lora_manager": {
+                    "enabled": True,
+                    "path": SEEDED_PATHS[0],
+                    "civitai_api_key": KEY,
+                }
+            }
+        )
+        GalleryConfig.MONITORING_DIRECTORIES = list(SEEDED_PATHS[1:3]) + [
+            SEEDED_PATHS[4]
+        ]
+        PromptManagerConfig.update_config(
+            {"database": {"default_path": SEEDED_PATHS[3]}}
+        )
+
+    def assert_no_paths_or_key(self, report, text):
+        self.assertNotIn(KEY, text)
+        self.assertNotIn(self.tmpdir, text)
+        self.assertNotIn(self.tmpdir.replace("\\", "/"), text)
+        for seeded in SEEDED_PATHS:
+            self.assertNotIn(seeded, text)
+            self.assertNotIn(json.dumps(seeded)[1:-1], text)
+        self.assertNotIn("someone", text)
+        for value in _strings(report):
+            self.assertFalse(posixpath.isabs(value), value)
+            self.assertFalse(ntpath.isabs(value), value)
+            self.assertIsNone(_WINDOWS_ABSOLUTE.match(value), value)
+            self.assertFalse(value.startswith("\\\\"), value)
+
+
+class TestReport(DiagnosticsReportTestCase):
+    async def test_report_has_version_viewerjs_and_the_four_checks(self):
+        report, _text = await self.report()
+        self.assertEqual(set(report), TOP_KEYS)
+        self.assertTrue(report["success"])
+        with open(
+            os.path.join(os.path.dirname(os.path.dirname(__file__)), "pyproject.toml"),
+            encoding="utf-8",
+        ) as handle:
+            version = re.search(r'version\s*=\s*["\']([^"\']+)["\']', handle.read())
+        self.assertEqual(report["version"], version.group(1))
+        self.assertRegex(report["viewerjs"], r"^\d+\.\d+\.\d+$")
+        self.assertRegex(report["python"], r"^3\.\d+\.\d+")
+        self.assertEqual(set(report["platform"]), {"system", "release"})
+        self.assertEqual(
+            set(report["checks"]),
+            {"database", "dependencies", "comfyui_output", "image_monitor"},
+        )
+        self.assertEqual(report["checks"]["database"]["status"], "ok")
+        self.assertEqual(
+            set(report["sources"]),
+            {"configured", "effective", "watching", "unavailable"},
+        )
+        self.assertEqual(set(report["logging"]), {"config", "stats"})
+        self.assertIsNone(report["health"])
+        self.assertEqual(report["redacted"], 0)
+        self.assertRegex(report["generated_at"], r"^\d{4}-\d\d-\d\dT\d\d:\d\d:\d\d$")
+
+    async def test_report_has_no_secret_or_absolute_path(self):
+        self.seed_secrets()
+        with mock.patch.object(
+            admin.sqlite3,
+            "connect",
+            side_effect=sqlite3.DatabaseError("file is not a database"),
+        ):
+            report, text = await self.report()
+        self.assertEqual(report["checks"]["database"]["status"], "error")
+        self.assertTrue(report["config"]["integrations"]["has_civitai_api_key"])
+        self.assertTrue(report["config"]["integrations"]["lora_manager_enabled"])
+        self.assertEqual(report["sources"]["configured"], 3)
+        self.assert_no_paths_or_key(report, text)
+        self.assertEqual(report["redacted"], 0)
+
+    async def test_report_leaves_get_config_unchanged(self):
+        self.seed_secrets()
+        before = json.dumps(PromptManagerConfig.get_config(), sort_keys=True)
+        directories = GalleryConfig.MONITORING_DIRECTORIES
+        await self.report()
+        self.assertEqual(
+            json.dumps(PromptManagerConfig.get_config(), sort_keys=True), before
+        )
+        self.assertIs(GalleryConfig.MONITORING_DIRECTORIES, directories)
+        self.assertEqual(IntegrationConfig.CIVITAI_API_KEY, KEY)
+
+    def test_outside_path_reads_outside_comfyui(self):
+        inside = os.path.join(str(self.comfy_dir), "user", "__promptmanager", "logs")
+        self.assertEqual(report_path(inside), "user/__promptmanager/logs")
+        self.assertEqual(
+            report_path(os.path.join(str(self.outside_dir), "vito")), "outside ComfyUI"
+        )
+        self.assertEqual(report_path(""), "")
+        self.assertEqual(report_path(None), "")
+
+    def test_a_filesystem_root_anchor_never_makes_a_path_public(self):
+        # --output-directory /renders makes "/" an anchor; a path outside
+        # ComfyUI must still read "outside ComfyUI", never "home/<user>/...".
+        root = os.path.abspath(os.sep)
+        anchors = [str(self.comfy_dir), root]
+        with mock.patch.object(GalleryConfig, "path_anchors", return_value=anchors):
+            self.assertEqual(
+                report_path(os.path.join(str(self.outside_dir), "x")), OUTSIDE
+            )
+            self.assertEqual(
+                report_path(os.path.join(str(self.comfy_dir), "output")), "output"
+            )
+
+    def test_scrub_removes_the_home_folder_without_its_root(self):
+        home = os.path.expanduser("~")
+        relative = os.path.splitdrive(home)[1].lstrip("\\/").replace("\\", "/")
+        cleaned = scrub({"a": f"{relative}/area/prompts.db", "redacted": 0}, None)
+        self.assertEqual(cleaned, {"a": "[removed]", "redacted": 1})
+
+    def test_the_home_folder_is_matched_as_a_whole_path_part(self):
+        with mock.patch.object(
+            diagnostics_report.os.path, "expanduser", return_value="/root"
+        ):
+            cleaned = scrub(
+                {
+                    "a": "output/rootfiles",
+                    "b": "output/root/x",
+                    "c": "see /root/x",
+                    "redacted": 0,
+                },
+                None,
+            )
+        self.assertEqual(cleaned["a"], "output/rootfiles")
+        # The home folder's name is a whole path part here: removed.
+        self.assertEqual(cleaned["b"], "[removed]")
+        self.assertEqual(cleaned["c"], "[removed]")
+        self.assertEqual(cleaned["redacted"], 2)
+
+    async def test_report_reads_only_the_temporary_log_folder(self):
+        self.write("prompt_manager.log.1", b"x" * 1234)
+        report, _text = await self.report()
+        stats = report["logging"]["stats"]
+        self.assertEqual(stats["log_files_count"], 2)
+        self.assertEqual(
+            stats["total_log_size"],
+            1234 + os.path.getsize(self.path("prompt_manager.log")),
+        )
+
+    def test_an_ancestor_of_home_is_never_an_anchor_on_linux(self):
+        anchored = diagnostics_report.anchored_path
+        home = "/home/someone"
+        anchors = ["/home", "/opt/ComfyUI"]
+        self.assertEqual(
+            anchored("/home/someone/renders/a.png", anchors, home, posixpath), OUTSIDE
+        )
+        self.assertEqual(
+            anchored("/home/other/renders", anchors, home, posixpath), OUTSIDE
+        )
+        self.assertEqual(
+            anchored("/opt/ComfyUI/output", anchors, home, posixpath), "output"
+        )
+
+    def test_an_ancestor_of_home_is_never_an_anchor_on_windows(self):
+        anchored = diagnostics_report.anchored_path
+        home = ntpath.normcase("C:\\Users\\someone")
+        anchors = [ntpath.normcase(a) for a in ("C:\\Users", "C:\\", "D:\\ComfyUI")]
+        path = ntpath.normcase("C:\\Users\\someone\\renders\\a.png")
+        self.assertEqual(anchored(path, anchors, home, ntpath), OUTSIDE)
+        inside = ntpath.normcase("D:\\ComfyUI\\output\\x")
+        self.assertEqual(anchored(inside, anchors, home, ntpath), "output/x")
+
+    def test_an_anchor_that_is_home_names_no_user(self):
+        anchored = diagnostics_report.anchored_path
+        home = "/home/someone"
+        self.assertEqual(
+            anchored("/home/someone/ComfyUI/output", [home], home, posixpath),
+            "ComfyUI/output",
+        )
+        self.assertEqual(
+            anchored(
+                "/home/someone/ComfyUI/output",
+                [home, "/home/someone/ComfyUI"],
+                home,
+                posixpath,
+            ),
+            "output",
+        )
+
+    def test_the_home_folder_name_is_removed_only_from_paths(self):
+        with mock.patch.object(
+            diagnostics_report.os.path, "expanduser", return_value="/home/info"
+        ):
+            cleaned = scrub({"level": "INFO", "p": "logs/info/x", "redacted": 0}, None)
+        self.assertEqual(cleaned, {"level": "INFO", "p": "[removed]", "redacted": 1})
+
+    def test_scrub_removes_the_home_folder_name(self):
+        with mock.patch.object(
+            diagnostics_report.os.path, "expanduser", return_value="/home/someone"
+        ):
+            cleaned = scrub(
+                {"a": "someone/renders", "b": "someones", "redacted": 0}, None
+            )
+        self.assertEqual(cleaned["a"], "[removed]")
+        self.assertEqual(cleaned["b"], "someones")
+
+    def test_scrub_removes_a_leaked_path_and_counts_it(self):
+        leaked = {
+            "a": "/home/someone/x",
+            "b": ["C:\\Users\\someone", "D:/data", "\\\\nas\\share", "ok"],
+            "c": {"d": f"token {KEY} here", "e": 3, "f": None, "g": "relative/path"},
+            "redacted": 0,
+        }
+        cleaned = scrub(leaked, KEY)
+        self.assertEqual(cleaned["redacted"], 5)
+        self.assertEqual(cleaned["a"], "[removed]")
+        self.assertEqual(cleaned["b"], ["[removed]"] * 3 + ["ok"])
+        self.assertEqual(
+            cleaned["c"], {"d": "[removed]", "e": 3, "f": None, "g": "relative/path"}
+        )
+        # A short key is not searched for (it would match ordinary words).
+        self.assertEqual(scrub({"x": "abc", "redacted": 0}, "abc")["redacted"], 0)
+
+    async def test_report_refused_while_exclusive_job_runs(self):
+        async with self.api._jobs.hold(
+            "restore", "Database restore", admission="exclusive"
+        ) as restore:
+            report, _text = await self.report(status=409)
+        self.assertEqual(report["code"], "busy")
+        self.assertEqual(
+            report["running"], {"id": restore.id, "title": "Database restore"}
+        )
+
+    async def test_report_lists_config_only_keys(self):
+        self.seed_secrets()
+        report, _text = await self.report()
+        self.assertEqual({k: set(v) for k, v in report["config"].items()}, CONFIG)
+        self.assertNotIn("default_path", json.dumps(report["config"]))
+        self.assertNotIn("directories", json.dumps(report["config"]))
+        self.assertNotIn(
+            "civitai_api_key",
+            json.dumps(report["config"]).replace("has_civitai_api_key", ""),
+        )
+
+    async def test_report_logging_has_public_directory(self):
+        report, _text = await self.report()
+        stats = report["logging"]["stats"]
+        self.assertFalse(posixpath.isabs(stats["log_directory"]))
+        self.assertIn("level", report["logging"]["config"])
+
+    async def test_report_survives_an_unreadable_viewerjs_provenance(self):
+        with mock.patch.object(
+            diagnostics_report,
+            "VIEWERJS_PROVENANCE",
+            os.path.join(self.tmpdir, "nope.json"),
+        ):
+            report, _text = await self.report()
+        self.assertEqual(report["viewerjs"], "unknown")
+
+    async def test_a_build_failure_is_coded_without_details(self):
+        with mock.patch.object(
+            self.api, "build_report", side_effect=RuntimeError(self.tmpdir)
+        ):
+            with self.assertLogs("prompt_manager", "ERROR"):
+                report, text = await self.report(status=500)
+        self.assertNotIn(self.tmpdir, text)
+        self.assertFalse(report["success"])
+
+
+class TestLegacyDiagnostics(DiagnosticsReportTestCase):
+    async def test_legacy_diagnostics_keeps_its_four_checks(self):
+        response = await self.client.request("GET", "/prompt_manager/diagnostics")
+        data = await response.json()
+        self.assertEqual(response.status, 200)
+        self.assertEqual(set(data), {"success", "diagnostics"})
+        self.assertEqual(
+            set(data["diagnostics"]),
+            {"database", "dependencies", "comfyui_output", "image_monitor"},
+        )
+        report, _text = await self.report()
+        self.assertEqual(report["checks"]["database"], data["diagnostics"]["database"])
+        self.assertEqual(
+            report["checks"]["dependencies"], data["diagnostics"]["dependencies"]
+        )
+
+    async def test_both_routes_build_their_checks_with_one_helper(self):
+        with mock.patch.object(
+            admin, "legacy_checks", wraps=admin.legacy_checks
+        ) as spy:
+            await self.client.request("GET", "/prompt_manager/diagnostics")
+            await self.report()
+        self.assertEqual(spy.call_count, 2)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_api_images.py

@@ -9,11 +9,9 @@ import asyncio
 import json
 import os
 import pathlib
-import shutil
 import sys
 import tempfile
 import threading
-import time
 import types
 import unittest
 from contextlib import ExitStack
@@ -38,8 +36,6 @@ from PIL import Image  # noqa: E402
 from database.operations import PromptDatabase  # noqa: E402
 import py.api.images as images_module
 from py.api import gallery_listing  # noqa: E402
-from py import lora_utils  # noqa: E402
-from py import lora_manager_layout as layout_module  # noqa: E402
 from py.api.thumbnail_settings import ThumbnailSettings  # noqa: E402
 from py.api import PromptManagerAPI  # noqa: E402
 from py.api.prompts import MAX_PAGE_LIMIT  # noqa: E402
@@ -115,25 +111,13 @@ class ImageAPITestCase(AioHTTPTestCase):
         self.assertGreater(image_id, 0, "fixture: image link failed")
         return image_id
 
-    def _make_symlink(self, target, link, target_is_directory=False):
+    def _make_symlink(self, target, link):
         """Create a symlink or skip the test when the platform refuses."""
         try:
-            os.symlink(str(target), str(link), target_is_directory=target_is_directory)
+            os.symlink(str(target), str(link))
         except (OSError, NotImplementedError, AttributeError) as exc:
             self.skipTest(f"symlinks unavailable: {exc}")
 
-    def _spy_executor(self):
-        """Record the __name__ of every callable sent to _run_in_executor."""
-        original = self.api._run_in_executor
-        calls = []
-
-        async def spy(func, *args, **kwargs):
-            calls.append(getattr(func, "__name__", repr(func)))
-            return await original(func, *args, **kwargs)
-
-        self.api._run_in_executor = spy
-        return calls
-
 
 class TestServeOutputImage(ImageAPITestCase):
     """GET /prompt_manager/images/serve/{filepath}"""
@@ -404,6 +388,18 @@ class ThumbnailTestCase(ImageAPITestCase):
 
     PROGRESS_URL = "/prompt_manager/images/generate-thumbnails/progress"
 
+    def _spy_executor(self):
+        """Record the __name__ of every callable sent to _run_in_executor."""
+        original = self.api._run_in_executor
+        calls = []
+
+        async def spy(func, *args, **kwargs):
+            calls.append(getattr(func, "__name__", repr(func)))
+            return await original(func, *args, **kwargs)
+
+        self.api._run_in_executor = spy
+        return calls
+
     def _thumb_path(self, name, suffix=".webp"):
         return self.output_dir / "thumbnails" / f"{name}_thumb{suffix}"
 
@@ -1220,13 +1216,6 @@ def _raise(*_args, **_kwargs):
 class ImageRouteCoverageCase(ImageAPITestCase):
     """Helpers for driving success and failure branches of each route."""
 
-    def setUp(self):
-        # #362: the LoraManager layout cache is per process; every test here
-        # starts and ends with it empty (runs before the subclasses' setUpAsync).
-        super().setUp()
-        lora_utils.clear_layout_cache()
-        self.addCleanup(lora_utils.clear_layout_cache)
-
     def _break_db(self, db_method):
         setattr(self.api.db, db_method, _raise)
 
@@ -1338,9 +1327,6 @@ class TestServeErrorBranches(ImageRouteCoverageCase):
                 )
                 self.assertEqual(resp.status, 200)
 
-            # The cache does not watch default_lora_roots (#362): without a
-            # clear the next request would reuse the folders above.
-            lora_utils.clear_layout_cache()
             with (
                 patch.object(IntegrationConfig, "LORA_MANAGER_ENABLED", True),
                 patch.object(IntegrationConfig, "LORA_MANAGER_PATH", ""),
@@ -1433,39 +1419,32 @@ class TestLoraManagerMediaServing(ImageRouteCoverageCase):
         self.cache = root / "cache"
         for folder in (self.sidecars, self.examples, self.cache):
             folder.mkdir()
-        self.root = root
         self.write_settings(str(self.sidecars))
         self.enabled = patch.object(IntegrationConfig, "LORA_MANAGER_ENABLED", True)
-        self.default_roots = patch("py.lora_utils.default_lora_roots", return_value=[])
         for item in (
             self.enabled,
             patch.object(IntegrationConfig, "LORA_MANAGER_PATH", str(self.lm)),
             patch(
                 "py.lora_utils.custom_nodes_directories", return_value=[custom_nodes]
             ),
-            self.default_roots,
+            patch("py.lora_utils.default_lora_roots", return_value=[]),
             patch("py.lora_utils.get_lora_image_cache_dir", return_value=self.cache),
             patch.dict(os.environ, {"LORA_MANAGER_SETTINGS_DIR": str(self.settings)}),
         ):
             item.start()
             self.addCleanup(item.stop)
 
-    def write_settings(self, sidecar_root, examples=None):
-        path = self.settings / "settings.json"
-        path.write_text(
+    def write_settings(self, sidecar_root):
+        (self.settings / "settings.json").write_text(
             json.dumps(
                 {
                     "sidecar_storage_mode": "centralized",
                     "sidecar_storage_path": sidecar_root,
-                    "example_images_path": str(examples or self.examples),
+                    "example_images_path": str(self.examples),
                     "civitai_api_key": "never-served",
                 }
             )
         )
-        # Ten seconds old, so the layout cache (real clock here) stores the
-        # first load instead of treating the file as being written (#362).
-        stamp = time.time_ns() - 10 * 10**9
-        os.utime(path, ns=(stamp, stamp))
 
     async def _status(self, image_id):
         resp = await self.client.request(
@@ -1485,65 +1464,11 @@ class TestLoraManagerMediaServing(ImageRouteCoverageCase):
 
         for image_id in ids:
             self.assertEqual(await self._status(image_id), 200)
-        # A warm layout cache does not outlive the integration (#362).
-        self.assertIsNotNone(lora_utils._layout_cache._entry)
         self.enabled.stop()
         for image_id in ids:
             self.assertEqual(await self._status(image_id), 403)
         self.enabled.start()
 
-    def _example(self, folder=None, name="1.mp4"):
-        path = (folder or self.examples) / "ab12" / name
-        path.parent.mkdir(exist_ok=True)
-        path.write_bytes(b"video")
-        return path
-
-    async def test_unchanged_settings_reuse_the_cached_layout(self):
-        image_id = self._link_image(self._example())
-        self.assertEqual(await self._status(image_id), 200)
-        self.assertIsNotNone(lora_utils._layout_cache._entry)
-        self.assertEqual(await self._status(image_id), 200)
-        self.assertEqual(lora_utils.default_lora_roots.call_count, 1)
-
-    async def test_a_settings_change_is_seen_without_a_clear(self):
-        image_id = self._link_image(self._example())
-        self.assertEqual(await self._status(image_id), 200)
-        self.assertIsNotNone(lora_utils._layout_cache._entry)
-        moved = self.root / "examples-moved"
-        moved.mkdir()
-        self.write_settings(str(self.sidecars), examples=moved)
-        self.assertEqual(await self._status(image_id), 403)
-
-    async def test_a_cached_folder_swapped_for_a_symlink_to_home_is_refused(self):
-        # CUJ 7: the example-images folder becomes a symlink to a folder that
-        # holds the home folder; the same recorded file is refused.
-        outer = self.root / "outer"
-        (outer / "home").mkdir(parents=True)
-        self._example(outer)
-        image_id = self._link_image(self._example())
-        with patch.object(layout_module.Path, "home", return_value=outer / "home"):
-            self.assertEqual(await self._status(image_id), 200)
-            self.assertIsNotNone(lora_utils._layout_cache._entry)
-            # A skip after the rmtree is safe: setUp builds a fresh tree.
-            shutil.rmtree(self.examples)
-            self._make_symlink(outer, self.examples, target_is_directory=True)
-            self.assertEqual(await self._status(image_id), 403)
-
-    async def test_a_configured_path_outside_custom_nodes_adds_nothing(self):
-        from py.config import IntegrationConfig
-
-        self.assertEqual(await self._status(self._link_image(self._example())), 200)
-        self.assertIsNotNone(lora_utils._layout_cache._entry)
-        with patch.object(IntegrationConfig, "LORA_MANAGER_PATH", str(self.sidecars)):
-            self.assertEqual(self.api._lora_image_dirs(), [])
-        # The warm entry for the real install was never handed out for it.
-        self.assertIsNone(lora_utils.cached_lora_manager_layout(str(self.sidecars))[0])
-
-    async def test_serve_image_builds_the_allow_list_in_the_executor(self):
-        calls = self._spy_executor()
-        self.assertEqual(await self._status(self._link_image(self._example())), 200)
-        self.assertIn("_lora_image_dirs", calls)
-
     async def test_a_filesystem_root_as_sidecar_folder_adds_nothing(self):
         self.write_settings(os.path.abspath(os.sep))
         dirs = self.api._lora_image_dirs()

tests/test_api_log_cleanup.py

@@ -0,0 +1,419 @@
+"""Reviewed rotated-log cleanup as a maintenance plan (#329, py/api/log_cleanup.py).
+
+``POST /maintenance/preview {operation: "rotated_logs"}`` lists exactly the
+rotated copies Rule R allows and writes nothing; ``POST /maintenance/apply``
+deletes exactly those as a ``maintenance`` job (admission ``read``), or
+nothing, failing ``plan-changed``, when any of them changed after the review.
+
+Every test runs on a real ``PromptManagerLogger`` whose log folder is a
+temporary folder, never the real one.
+"""
+
+import asyncio
+import json
+import logging
+import ntpath
+import os
+import posixpath
+import re
+import sys
+import threading
+import unittest
+import unittest.mock as mock
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+from py.api.maintenance_plans import PlanStore  # noqa: E402
+from utils import logging_config  # noqa: E402
+from utils.logging_config import PromptManagerLogger, get_logger_manager  # noqa: E402
+
+try:
+    from tests.open_handles import assert_closed  # noqa: E402
+    from tests.test_api_maintenance import MaintenanceAPITestCase  # noqa: E402
+except ImportError:  # discovered with tests/ as the top-level directory
+    from open_handles import assert_closed  # noqa: E402
+    from test_api_maintenance import MaintenanceAPITestCase  # noqa: E402
+
+WAIT = 10
+PREVIEW = "/prompt_manager/maintenance/preview"
+APPLY = "/prompt_manager/maintenance/apply"
+CHANGED = "The log files changed since the review; nothing was deleted. Review again."
+MEDIA = "Deletes only the listed log files. No image, video or library record changes."
+_WINDOWS_ABSOLUTE = re.compile(r"^[A-Za-z]:[\\/]")
+
+
+def _strings(value):
+    if isinstance(value, str):
+        yield value
+    elif isinstance(value, dict):
+        for item in value.values():
+            yield from _strings(item)
+    elif isinstance(value, (list, tuple)):
+        for item in value:
+            yield from _strings(item)
+
+
+class LogCleanupTestCase(MaintenanceAPITestCase):
+    async def get_application(self):
+        app = await super().get_application()
+        package = os.path.join(self.tmpdir, "pkg")
+        os.makedirs(os.path.join(package, "utils"))
+        fake_file = os.path.join(package, "utils", "logging_config.py")
+        with mock.patch.object(logging_config, "__file__", fake_file):
+            self.manager = object.__new__(PromptManagerLogger)  # bypass singleton
+            self.manager.__init__()
+        self.addCleanup(self._restore_global_logging)
+        self.manager.update_config({"console_logging": False})
+        self._only_this_thread()
+        self.log_dir = str(self.manager.log_dir)
+        self.app_logger = logging.getLogger("prompt_manager.log_cleanup_test")
+        self.app_logger.info("active log line")
+        self.api._get_logger_manager = lambda: self.manager
+        self.outside = os.path.join(self.tmpdir, "link-targets")
+        os.makedirs(self.outside)
+        return app
+
+    def _only_this_thread(self):
+        """Only this test's records reach the temporary log (not other threads')."""
+        this_thread = threading.get_ident()
+        for handler in self.manager.logger.handlers:
+            handler.filters.clear()
+            handler.addFilter(lambda record: record.thread == this_thread)
+
+    def _restore_global_logging(self):
+        for handler in logging.getLogger("prompt_manager").handlers[:]:
+            handler.close()
+        assert_closed(self, self.log_dir)
+        get_logger_manager()._setup_loggers()
+
+    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 symlink(self, name):
+        target = os.path.join(self.outside, "secret.txt")
+        with open(target, "wb") as handle:
+            handle.write(b"secret " * 100)
+        try:
+            os.symlink(target, self.path(name))
+        except (OSError, NotImplementedError):  # pragma: no cover - no privilege
+            return False
+        return True
+
+    def names(self):
+        return sorted(os.listdir(self.log_dir))
+
+    def snapshot(self):
+        """Every entry but the active log, with its bytes (None for a folder)."""
+        state = {}
+        for name in self.names():
+            if name == "prompt_manager.log":
+                continue
+            path = self.path(name)
+            if os.path.islink(path):
+                state[name] = ("link", os.readlink(path))
+            elif os.path.isdir(path):
+                state[name] = None
+            else:
+                with open(path, "rb") as handle:
+                    state[name] = handle.read()
+        return state
+
+    def seed(self):
+        """CUJ 5's folder: three rotated copies, a .bak, a folder and a link."""
+        self.write("prompt_manager.log.1", b"a" * 1000)
+        self.write("prompt_manager.log.2", b"b" * 2000)
+        self.write("prompt_manager.log.3", b"c" * 3000)
+        self.write("prompt_manager.log.bak", b"mine")
+        os.makedirs(self.path("prompt_manager.log.9"))
+        return self.symlink("prompt_manager.log.4")
+
+    async def preview(self, status=200, **body):
+        return await self.post(
+            PREVIEW, {"operation": "rotated_logs", "options": {}, **body}, status
+        )
+
+    async def apply(self, token, status=202):
+        return await self.post(APPLY, {"token": token}, status)
+
+    async def run_apply(self, plan):
+        body = await self.apply(plan["token"])
+        job = self.api._jobs.get(body["job"]["id"])
+        await asyncio.wait_for(asyncio.shield(job.task), WAIT)
+        return body["job"], job
+
+
+class TestPreview(LogCleanupTestCase):
+    async def test_preview_lists_only_rotated_copies(self):
+        linked = self.seed()
+        plan = await self.preview()
+        self.assertEqual(
+            set(plan),
+            {
+                "success",
+                "token",
+                "operation",
+                "expires_at",
+                "scope",
+                "protected",
+                "backup",
+                "media",
+                "details",
+            },
+        )
+        self.assertEqual(plan["operation"], "rotated_logs")
+        self.assertEqual(
+            plan["scope"], {"count": 3, "label": "3 rotated log copies · 5.9 KB"}
+        )
+        self.assertEqual(
+            [row["label"] for row in plan["details"]],
+            ["prompt_manager.log.1", "prompt_manager.log.2", "prompt_manager.log.3"],
+        )
+        self.assertRegex(
+            plan["details"][2]["detail"], r"^2\.9 KB · \d{4}-\d\d-\d\d \d\d:\d\d$"
+        )
+        reasons = ["Kept: the active log"]
+        if linked:
+            reasons.append("Kept: 1 link")
+        reasons.append("Kept: 2 other files")
+        self.assertEqual(
+            plan["protected"], {"count": len(reasons) + 1, "reasons": reasons}
+        )
+        self.assertEqual(
+            plan["backup"],
+            {"policy": "none", "destination_label": None, "expected_bytes": None},
+        )
+        self.assertEqual(plan["media"], {"deletes_files": True, "label": MEDIA})
+        self.assertTrue(plan["expires_at"].endswith("+00:00"))
+        self.assertNotIn("mtime_ns", json.dumps(plan))
+
+    async def test_preview_writes_nothing(self):
+        self.seed()
+        before = self.snapshot()
+        await self.preview()
+        self.assertEqual(self.snapshot(), before)
+
+    async def test_preview_has_no_paths(self):
+        self.seed()
+        plan = await self.preview()
+        for text in _strings(plan):
+            self.assertFalse(posixpath.isabs(text), text)
+            self.assertFalse(ntpath.isabs(text), text)
+            self.assertIsNone(_WINDOWS_ABSOLUTE.match(text), text)
+
+    async def test_preview_rejects_non_empty_options(self):
+        for options in ({"all": True}, [], "x"):
+            with self.subTest(options=options):
+                reply = await self.preview(status=400, options=options)
+                self.assertEqual(reply["code"], "invalid-request")
+
+    async def test_preview_too_large_is_413(self):
+        self.seed()
+        self.api._maintenance_plans = PlanStore(max_bytes=10)
+        reply = await self.preview(status=413)
+        self.assertEqual(
+            reply,
+            {
+                "success": False,
+                "code": "too-large",
+                "error": "Too many log files to review at once.",
+            },
+        )
+
+    async def test_empty_folder_plan_has_scope_zero(self):
+        plan = await self.preview()
+        self.assertEqual(plan["scope"], {"count": 0, "label": "No rotated log copies"})
+        self.assertEqual(plan["details"], [])
+        self.assertEqual(
+            plan["protected"], {"count": 1, "reasons": ["Kept: the active log"]}
+        )
+
+    async def test_preview_refused_while_an_exclusive_job_runs(self):
+        async with self.api._jobs.hold(
+            "restore", "Database restore", admission="exclusive"
+        ):
+            reply = await self.preview(status=409)
+        self.assertEqual(reply["code"], "busy")
+
+
+class TestApply(LogCleanupTestCase):
+    async def test_apply_deletes_exactly_the_reviewed_files(self):
+        linked = self.seed()
+        plan = await self.preview()
+        summary, job = await self.run_apply(plan)
+        self.assertEqual(summary["kind"], "maintenance")
+        self.assertEqual(summary["title"], "Rotated log copies")
+        self.assertEqual(summary["admission"], "read")
+        self.assertFalse(summary["cancellable"])
+        self.assertEqual(summary["scope"], {"operation": "rotated_logs"})
+        self.assertEqual(job.state, "finished", job.log)
+        self.assertEqual(
+            job.counters,
+            {"changed": 3, "protected": plan["protected"]["count"], "errors": 0},
+        )
+        expected = [
+            "prompt_manager.log",
+            "prompt_manager.log.9",
+            "prompt_manager.log.bak",
+        ]
+        if linked:
+            expected.append("prompt_manager.log.4")
+        self.assertEqual(self.names(), sorted(expected))
+        self.assertEqual(job.status_line, "Deleted 3 rotated log copies")
+        report = json.dumps(job.report())
+        self.assertNotIn(self.tmpdir, report)
+
+    async def test_apply_rows_name_each_file(self):
+        self.seed()
+        plan = await self.preview()
+        _summary, job = await self.run_apply(plan)
+        self.assertEqual(
+            job.outcomes["changed"],
+            [
+                {
+                    "path": "prompt_manager.log.1",
+                    "root": "",
+                    "root_index": None,
+                    "detail": "Deleted · 1000 bytes",
+                },
+                {
+                    "path": "prompt_manager.log.2",
+                    "root": "",
+                    "root_index": None,
+                    "detail": "Deleted · 2.0 KB",
+                },
+                {
+                    "path": "prompt_manager.log.3",
+                    "root": "",
+                    "root_index": None,
+                    "detail": "Deleted · 2.9 KB",
+                },
+            ],
+        )
+        log = "\n".join(job.log)
+        for n in (1, 2, 3):
+            self.assertIn(f"prompt_manager.log.{n}", log)
+
+    async def test_apply_after_rollover_fails_plan_changed_and_deletes_nothing(self):
+        self.seed()
+        plan = await self.preview()
+        handler = next(
+            h
+            for h in self.manager.logger.handlers
+            if isinstance(h, logging.FileHandler)
+        )
+        handler.acquire()
+        try:
+            handler.doRollover()  # every listed name now holds another file
+        finally:
+            handler.release()
+        before = self.snapshot()
+        _summary, job = await self.run_apply(plan)
+        self.assertEqual(job.state, "failed")
+        self.assertEqual(job.scope["failure"], "plan-changed")
+        self.assertEqual(job.status_line, CHANGED)
+        self.assertEqual(job.counters["changed"], 0)
+        self.assertEqual(self.snapshot(), before)
+
+    async def test_apply_reports_a_file_in_use(self):
+        self.seed()
+        plan = await self.preview()
+        real_unlink = os.unlink
+        held = self.path("prompt_manager.log.2")
+
+        def unlink(path, *args, **kwargs):
+            if os.fspath(path) == held:
+                error = PermissionError(
+                    13,
+                    "The process cannot access the file because it is being "
+                    "used by another process",
+                    held,
+                )
+                error.winerror = 32
+                raise error
+            return real_unlink(path, *args, **kwargs)
+
+        with mock.patch.object(logging_config.os, "unlink", unlink):
+            _summary, job = await self.run_apply(plan)
+        self.assertEqual(job.counters["changed"], 2)
+        self.assertEqual(job.counters["errors"], 1)
+        [row] = job.outcomes["errors"]
+        self.assertEqual(row["path"], "prompt_manager.log.2")
+        self.assertIn("being used by another process", row["detail"])
+        self.assertNotIn(self.tmpdir, json.dumps(job.report()))
+        self.assertIn("prompt_manager.log.2", self.names())
+        self.assertEqual(
+            job.status_line, "Deleted 2 rotated log copies, 1 could not be deleted"
+        )
+
+    async def test_token_is_single_use(self):
+        self.seed()
+        plan = await self.preview()
+        summary, _job = await self.run_apply(plan)
+        again = await self.apply(plan["token"], 409)
+        self.assertEqual(again["code"], "plan-used")
+        self.assertEqual(
+            again["job"], {"id": summary["id"], "title": "Rotated log copies"}
+        )
+
+    async def test_apply_runs_beside_a_scan(self):
+        self.seed()
+        plan = await self.preview()
+        async with self.api._jobs.hold("scan", "Metadata scan", admission="records"):
+            _summary, job = await self.run_apply(plan)
+        self.assertEqual(job.state, "finished", job.log)
+        self.assertNotIn("prompt_manager.log.1", self.names())
+
+    async def test_apply_refused_while_restore_runs(self):
+        self.seed()
+        plan = await self.preview()
+        async with self.api._jobs.hold(
+            "restore", "Database restore", admission="exclusive"
+        ) as restore:
+            busy = await self.apply(plan["token"], 409)
+        self.assertEqual(busy["code"], "busy")
+        self.assertEqual(
+            busy["running"], {"id": restore.id, "title": "Database restore"}
+        )
+        self.assertIn("prompt_manager.log.1", self.names())
+        # The refused apply kept the token.
+        _summary, job = await self.run_apply(plan)
+        self.assertEqual(job.state, "finished", job.log)
+
+    async def test_apply_refused_beside_another_maintenance_job(self):
+        self.seed()
+        plan = await self.preview()
+        async with self.api._jobs.hold(
+            "maintenance", "Optimize database", admission="records"
+        ):
+            busy = await self.apply(plan["token"], 409)
+        self.assertEqual(busy["code"], "busy")
+
+    async def test_active_log_survives_and_keeps_growing(self):
+        self.seed()
+        plan = await self.preview()
+        await self.run_apply(plan)
+        self.app_logger.info("written after the cleanup")
+        with open(self.path("prompt_manager.log"), "rb") as handle:
+            content = handle.read()
+        self.assertIn(b"active log line", content)
+        self.assertIn(b"written after the cleanup", content)
+
+    async def test_an_unexpected_failure_fails_the_job_without_details(self):
+        self.seed()
+        plan = await self.preview()
+        with mock.patch.object(
+            self.manager, "delete_rotated", side_effect=RuntimeError(self.tmpdir)
+        ):
+            with self.assertLogs("prompt_manager", "ERROR"):
+                _summary, job = await self.run_apply(plan)
+        self.assertEqual(job.state, "failed")
+        self.assertNotIn(self.tmpdir, json.dumps(job.report()))
+        self.assertEqual(job.status_line, "The log cleanup failed.")
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_api_logging.py

@@ -5,9 +5,10 @@ import os
 import shutil
 import sys
 import tempfile
+import threading
 import unittest
 from pathlib import Path
-from unittest.mock import MagicMock
+from unittest.mock import MagicMock, patch
 
 _mock_server = MagicMock()
 _mock_server.PromptServer.instance.routes = MagicMock()
@@ -20,7 +21,9 @@ from aiohttp.test_utils import AioHTTPTestCase  # noqa: E402
 
 from database.operations import PromptDatabase  # noqa: E402
 from py.api import PromptManagerAPI  # noqa: E402
+from py.api import logging_routes  # noqa: E402
 from py.api.logging_routes import LOG_CONFIG_KEYS  # noqa: E402
+from utils import logging_config  # noqa: E402
 
 
 class FakeLoggerManager:
@@ -38,11 +41,16 @@ class FakeLoggerManager:
         }
         self.update_calls = []
         self.logs = []
+        self.threads = {}
 
     def update_config(self, new_config):
+        self.threads["update_config"] = threading.current_thread()
         self.update_calls.append(dict(new_config))
         self.config.update(new_config)
 
+    def log_file_entry(self, name):
+        return logging_config.log_file_entry(self.log_dir, name)
+
     def get_config(self):
         return dict(self.config)
 
@@ -54,23 +62,36 @@ class FakeLoggerManager:
 
     def get_log_files(self):
         files = []
-        for path in sorted(self.log_dir.glob("*.log*")):
+        for row in sorted(
+            logging_config.scan_log_folder(self.log_dir), key=lambda r: r["name"]
+        ):
+            if row["_dir"]:
+                continue
             files.append(
                 {
-                    "filename": path.name,
-                    "path": str(path),
-                    "size": path.stat().st_size,
+                    "filename": row["name"],
+                    "path": str(self.log_dir / row["name"]),
+                    "size": row["size"],
                     "modified": "2026-10-07T00:00:00",
-                    "is_main": path.name == "prompt_manager.log",
+                    "is_main": row["kind"] == "active",
+                    "kind": row["kind"],
                 }
             )
         return files
 
     def truncate_logs(self):
+        """Rule R: empty the active log, delete rotated copies, keep the rest."""
+        self.threads["truncate_logs"] = threading.current_thread()
         truncated = []
-        for path in self.log_dir.glob("prompt_manager.log*"):
-            path.write_text("")
-            truncated.append(path.name)
+        for row in logging_config.scan_log_folder(self.log_dir):
+            path = self.log_dir / row["name"]
+            if row["kind"] == "active":
+                path.write_text("")
+            elif row["kind"] == "rotated":
+                path.unlink()
+            else:
+                continue
+            truncated.append(row["name"])
         return {"truncated": truncated, "errors": []}
 
     def get_log_stats(self):
@@ -378,12 +399,129 @@ class TestLogFilesAndDownload(LoggingAPITestCase):
         self.assertEqual(resp.status, 400)
 
     async def test_download_read_failure_is_500(self):
-        (self.log_dir / "dir.log").mkdir()
-        resp = await self.client.request("GET", "/prompt_manager/logs/download/dir.log")
+        # Rewritten (#329): a directory is now a 404; inject a read error.
+        (self.log_dir / "prompt_manager.log").write_bytes(b"x")
+        secret = os.path.join(self.tmpdir, "logs", "prompt_manager.log")
+
+        def fail(path):
+            raise PermissionError(13, "Permission denied", secret)
+
+        with patch.object(logging_routes, "_read_log_bytes", fail):
+            resp = await self.client.request(
+                "GET", "/prompt_manager/logs/download/prompt_manager.log"
+            )
         body = await resp.text()
         self.assertEqual(resp.status, 500)
         self.assertNotIn(self.tmpdir, body)
 
+    async def test_directory_download_is_404(self):
+        (self.log_dir / "dir.log").mkdir()
+        resp = await self.client.request("GET", "/prompt_manager/logs/download/dir.log")
+        self.assertEqual(resp.status, 404)
+
+    async def test_download_refuses_a_symlink(self):
+        secret = Path(self.tmpdir) / "secret.txt"
+        secret.write_text("the key")
+        try:
+            os.symlink(secret, self.log_dir / "prompt_manager.log.3")
+        except (OSError, NotImplementedError):  # pragma: no cover
+            self.skipTest("this account cannot create symlinks")
+        resp = await self.client.request(
+            "GET", "/prompt_manager/logs/download/prompt_manager.log.3"
+        )
+        body = await resp.text()
+        self.assertEqual(resp.status, 404)
+        self.assertNotIn("the key", body)
+
+    async def test_download_refuses_a_hard_link(self):
+        secret = Path(self.tmpdir) / "secret.txt"
+        secret.write_text("the key")
+        try:
+            os.link(secret, self.log_dir / "prompt_manager.log.3")
+        except (OSError, NotImplementedError):  # pragma: no cover
+            self.skipTest("this filesystem cannot create hard links")
+        resp = await self.client.request(
+            "GET", "/prompt_manager/logs/download/prompt_manager.log.3"
+        )
+        self.assertEqual(resp.status, 404)
+        self.assertNotIn("the key", await resp.text())
+
+    async def test_download_refuses_a_file_swapped_after_it_was_listed(self):
+        (self.log_dir / "prompt_manager.log.2").write_text("listed")
+        other = Path(self.tmpdir) / "other.txt"
+        other.write_text("swapped in")
+        real_entry = self.manager.log_file_entry
+
+        def entry_then_swap(name):
+            path = real_entry(name)
+            os.replace(other, path)  # another file now has the listed name
+            return path
+
+        self.manager.log_file_entry = entry_then_swap
+        resp = await self.client.request(
+            "GET", "/prompt_manager/logs/download/prompt_manager.log.2"
+        )
+        self.assertEqual(resp.status, 404)
+        self.assertNotIn("swapped in", await resp.text())
+
+    @unittest.skipUnless(hasattr(os, "mkfifo"), "FIFOs are POSIX only")
+    async def test_download_refuses_a_fifo_without_blocking(self):
+        os.mkfifo(self.log_dir / "pipe.log")
+        resp = await self.client.request(
+            "GET", "/prompt_manager/logs/download/pipe.log"
+        )
+        self.assertEqual(resp.status, 404)
+
+    async def test_download_reads_off_the_loop(self):
+        (self.log_dir / "prompt_manager.log").write_bytes(b"line\n")
+        seen = []
+        real = logging_routes._read_log_bytes
+
+        def read(path):
+            seen.append(threading.current_thread())
+            return real(path)
+
+        with patch.object(logging_routes, "_read_log_bytes", read):
+            resp = await self.client.request(
+                "GET", "/prompt_manager/logs/download/prompt_manager.log"
+            )
+        self.assertEqual(resp.status, 200)
+        self.assertEqual(len(seen), 1)
+        self.assertIsNot(seen[0], threading.current_thread())
+
+    async def test_download_keeps_its_response_shape(self):
+        (self.log_dir / "prompt_manager.log.2").write_bytes(b"a\nb\n")
+        resp = await self.client.request(
+            "GET", "/prompt_manager/logs/download/prompt_manager.log.2"
+        )
+        self.assertEqual(resp.status, 200)
+        self.assertEqual(resp.content_type, "text/plain")
+        self.assertEqual(
+            resp.headers["Content-Disposition"],
+            'attachment; filename="prompt_manager.log.2"',
+        )
+        self.assertEqual(resp.headers["Content-Length"], "4")
+        self.assertEqual(await resp.read(), b"a\nb\n")
+
+    async def test_log_files_rows_carry_their_kind(self):
+        for name in (
+            "prompt_manager.log",
+            "prompt_manager.log.1",
+            "prompt_manager.log.bak",
+        ):
+            (self.log_dir / name).write_text("x")
+        resp = await self.client.request("GET", "/prompt_manager/logs/files")
+        data = await resp.json()
+        kinds = {row["filename"]: row["kind"] for row in data["files"]}
+        self.assertEqual(
+            kinds,
+            {
+                "prompt_manager.log": "active",
+                "prompt_manager.log.1": "rotated",
+                "prompt_manager.log.bak": "other",
+            },
+        )
+
 
 class TestTruncateConfigAndStats(LoggingAPITestCase):
 
@@ -395,6 +533,40 @@ class TestTruncateConfigAndStats(LoggingAPITestCase):
         self.assertEqual(data["results"]["truncated"], ["prompt_manager.log"])
         self.assertEqual((self.log_dir / "prompt_manager.log").read_text(), "")
 
+    async def test_truncate_keeps_other_files(self):
+        for name in (
+            "prompt_manager.log",
+            "prompt_manager.log.1",
+            "prompt_manager.log.bak",
+        ):
+            (self.log_dir / name).write_text("data")
+        resp = await self.client.request("POST", "/prompt_manager/logs/truncate")
+        data = await resp.json()
+        self.assertEqual(resp.status, 200)
+        self.assertEqual(set(data), {"success", "message", "results"})
+        self.assertEqual(
+            sorted(data["results"]["truncated"]),
+            ["prompt_manager.log", "prompt_manager.log.1"],
+        )
+        self.assertEqual(
+            sorted(p.name for p in self.log_dir.iterdir()),
+            ["prompt_manager.log", "prompt_manager.log.bak"],
+        )
+
+    async def test_truncate_runs_off_the_loop(self):
+        resp = await self.client.request("POST", "/prompt_manager/logs/truncate")
+        self.assertEqual(resp.status, 200)
+        self.assertIsNot(
+            self.manager.threads["truncate_logs"], threading.current_thread()
+        )
+
+    async def test_update_config_runs_off_the_loop(self):
+        resp = await self._post_config({"level": "DEBUG"})
+        self.assertEqual(resp.status, 200)
+        self.assertIsNot(
+            self.manager.threads["update_config"], threading.current_thread()
+        )
+
     async def test_truncate_failure_is_500(self):
         def boom():
             raise RuntimeError("locked")

tests/test_api_lora.py

@@ -876,18 +876,6 @@ class TestLoraScanRoute(LoraAPITestCase):
         resp = await self.client.request("POST", "/prompt_manager/lora/scan")
         self.assertEqual(resp.status, 400)
 
-    async def test_a_finished_scan_clears_the_layout_cache(self):
-        # #362: the legacy scan replaces through the same method.
-        self._enable()
-        loras = self.tmp_root / "loras"
-        loras.mkdir()
-        _write_metadata(loras, "plain")
-        (loras / "plain.png").write_bytes(_png_bytes())
-        with patch("py.lora_utils.clear_layout_cache") as clear:
-            resp, events, _calls = await self._scan(loras)
-        self.assertEqual(events[-1]["type"], "complete", events)
-        clear.assert_called_once_with()
-
     async def test_scan_imports_loras_with_images_and_links_existing(self):
         self._enable()
         loras = self.tmp_root / "loras"

tests/test_execution_history_hooks.py

@@ -356,8 +356,6 @@ class HistoryHooksTests(unittest.TestCase):
             self.assertGreaterEqual(time.monotonic() - began, 1.8)
             self.assertLess(time.monotonic() - began, CEILING_SECONDS)
             release.set()
-            # Drain first: the 2 s bound is not under test here (#370).
-            self.drain(registration)
             self.assertTrue(registration.dispose(2))
 
     def test_restore_disables_capture_and_clears_handles_before_fresh_epoch(self):

tests/test_execution_output_hooks.py

@@ -399,10 +399,6 @@ class OutputHooksTests(fixtures.unittest.TestCase):
             self.queue.task_done(1, history("rejected.png"), fixtures.status())
             registration.writer.invalidate()
             release.set()
-            # Let the writer work through the 1,024 queued commands first
-            # (#370's pattern): on a busy Windows runner that took longer
-            # than dispose's 2 s bound (merge queue, #382).
-            self.drain(registration)
             self.assertTrue(registration.dispose(2))
         self.assertEqual(self.outputs(0)["state"], "not_captured")
         self.assertEqual(len(self.rows()), 1)

tests/test_execution_output_writer.py

@@ -28,11 +28,6 @@ except ImportError:
 # Upper bound a blocked fake write waits for its release. Tests prove ordering
 # with events, so this only caps a failing run and never decides a pass.
 BLOCK_SECONDS = 10
-# A restore is a full verify/migrate/backup/replace cycle and is bound by the
-# disk: a busy Windows runner took over 10 s (merge queue, #382). Waits that
-# span a restore use this ceiling, which stays under the runner's 120 s hang
-# watchdog, so a real hang still fails with the thread dump.
-RESTORE_SECONDS = 90
 
 
 class OutputWriterTests(fixtures.unittest.TestCase):
@@ -322,8 +317,8 @@ class OutputWriterTests(fixtures.unittest.TestCase):
 
         Returns the patch and an event set once that rotation has returned on
         the worker. A restore is a full verify/migrate/backup/replace cycle, so
-        the handoff waits on this event with the ``RESTORE_SECONDS`` ceiling
-        instead of the short worker-behavior deadline (TESTING.md,
+        the handoff waits on this event with the generous ``BLOCK_SECONDS``
+        ceiling instead of the short worker-behavior deadline (TESTING.md,
         Windows portability preflight).
         """
         backup = os.path.join(self.directory.name, f"restore-{start.run_id}.db")
@@ -366,7 +361,7 @@ class OutputWriterTests(fixtures.unittest.TestCase):
                         )
                         with rotation:
                             self.finalization(writer, start)
-                            self.assertTrue(rotated.wait(RESTORE_SECONDS))
+                            self.assertTrue(rotated.wait(BLOCK_SECONDS))
                             self.wait_for(lambda: writer.status().pending == 0)
                         self.assertEqual(
                             self.db.get_execution_outputs(start.run_id)["state"],
@@ -408,7 +403,7 @@ class OutputWriterTests(fixtures.unittest.TestCase):
             self.assertEqual(writer.status().pending, 1)
             self.assertEqual(writer.context_budget.charged_bytes, charge)
             release.set()
-            self.assertTrue(rotated.wait(RESTORE_SECONDS))
+            self.assertTrue(rotated.wait(BLOCK_SECONDS))
             self.wait_for(lambda: writer.status().pending == 0)
         self.assertEqual(writer.context_budget.charged_bytes, 0)
         self.assertFalse(writer.status().accepting)
@@ -572,8 +567,6 @@ class OutputWriterTests(fixtures.unittest.TestCase):
                 self.assertIs(replacement.context_budget, writer.context_budget)
                 self.assertIsNone(replacement.context_budget.reserve(8388608))
                 release.set()
-                # Drain first: the 2 s bound is not under test here (#370).
-                drain_writer(writer)
                 self.assertTrue(writer.close(2))
         finally:
             release.set()
@@ -615,8 +608,6 @@ class OutputWriterTests(fixtures.unittest.TestCase):
             writer.invalidate()
             self.assertEqual(writer.context_budget.charged_bytes, 0)
             release.set()
-        # Drain first: the 2 s bound is not under test here (#370).
-        drain_writer(writer)
         self.assertTrue(writer.close(2))
 
     def test_malformed_context_omission_preserves_independent_valid_output(self):
@@ -707,8 +698,6 @@ class OutputWriterTests(fixtures.unittest.TestCase):
                 release.set()
                 invalidator.join(BLOCK_SECONDS)
             self.assertFalse(invalidator.is_alive())
-            # Drain first: the 2 s bound is not under test here (#370).
-            drain_writer(writer)
             self.assertTrue(writer.close(2))
         self.assertEqual(
             self.db.get_execution_outputs(start.run_id)["state"], "complete"

tests/test_image_monitor_singleton.py

@@ -365,39 +365,11 @@ class TestStartStop(unittest.TestCase):
     def test_stop_monitoring_reports_an_observer_stop_that_raised(self):
         fake = FakeObserver(stop_error=OSError("descriptor gone"))
         self._start_with(fake)
-        fake.stopped = True  # its thread has ended; stop() still raises
         with self.assertLogs("prompt_manager.image_monitor", "WARNING") as logs:
             self.assertFalse(self.monitor.stop_monitoring())
-        text = "\n".join(logs.output)
-        self.assertIn("Error while stopping image monitoring", text)
-        self.assertNotIn("watchdog observer still running", text)
+        self.assertIn("Error while stopping image monitoring", "\n".join(logs.output))
         fake.stop_error = None  # tearDown's stop calls stop() again
 
-    def test_the_still_running_warning_names_the_time_the_join_got(self):
-        fake = FakeObserver(stay_alive=True)
-        handler = self._start_with(fake)
-        now = [100.0]
-        recorded_close = handler.close
-
-        def close_that_uses_the_deadline(timeout=im_mod.WORKER_JOIN_SECONDS):
-            ok = recorded_close(timeout)
-            now[0] += 5.0  # a busy worker took the whole deadline
-            return ok
-
-        handler.close = close_that_uses_the_deadline
-        with (
-            patch.object(
-                im_mod, "time", types.SimpleNamespace(monotonic=lambda: now[0])
-            ),
-            self.assertLogs("prompt_manager.image_monitor", "WARNING") as logs,
-        ):
-            self.assertFalse(self.monitor.stop_monitoring(timeout=5.0))
-        self.assertEqual(fake.calls[-1], ("join", 0.0))
-        self.assertIn(
-            "watchdog observer still running after 0.0 s", "\n".join(logs.output)
-        )
-        fake.stay_alive = False  # it ends; tearDown's stop waits for it
-
     def test_stop_monitoring_without_an_observer_returns_true(self):
         self.assertTrue(self.monitor.stop_monitoring())
 

tests/test_init_package.py

@@ -603,24 +603,6 @@ def hooked_app():
     )
 
 
-WAIT_SECONDS = 30  # ceiling for every Event wait (TESTING.md test rules)
-
-
-def start_busy_worker(package, folder, name="busy.png"):
-    """The image worker alive in its processing delay."""
-    handler = package._global_image_monitor.handler
-    if handler is None:
-        raise AssertionError("the package's monitor did not start")
-    handler.processing_delay = 3600
-    # Outside the watched folder, so the observer does not schedule it first.
-    path = os.path.join(folder, name)
-    with open(path, "wb") as image:
-        image.write(b"x" * 16)
-    if not handler.schedule(path) or not handler._worker.is_alive():
-        raise AssertionError("the image worker did not start")
-    return handler._worker
-
-
 class PackageTeardownCase(unittest.TestCase):
     """Each test loads its own package (#301): the teardown runs once."""
 
@@ -649,83 +631,54 @@ class PackageTeardownCase(unittest.TestCase):
         return package
 
     def busy_worker(self, package, name="busy.png"):
-        return start_busy_worker(package, self.tmp.name, name)
-
-    def hold_a_preview(self, package, *, advance, kill_releases):
-        """A preview worker held inside a fake encode that tracks a fake
-        ffmpeg, and a package clock that the monitor step moves by
-        ``advance`` seconds. Cleanups release the encode first."""
-        # Not the watched folder. Real path for both the file and the root:
-        # macOS /var vs /private/var, Windows short names.
-        previews = os.path.realpath(os.path.join(self.tmp.name, "previews"))
-        os.makedirs(previews)
-        from PIL import Image
-
-        image = os.path.join(previews, "held.png")
-        Image.new("RGB", (8, 8), "red").save(image)
-        held = types.SimpleNamespace(
-            entered=threading.Event(),
-            release=threading.Event(),
-            killed=threading.Event(),
-            stop_seen=[],
-            timeouts=[],
-            now=[100.0],
-        )
-
-        class FakeFfmpeg:
-            def poll(self):  # _track keeps only processes still running
-                return None
-
-            def kill(self):
-                held.killed.set()
-                if kill_releases:
-                    held.release.set()
-
-        def held_encode(src, dst, settings, is_video, *, stop, guard, track):
-            track(FakeFfmpeg())
-            held.entered.set()
-            held.release.wait(WAIT_SECONDS)
-            held.stop_seen.append(stop())
-            return {"action": "skipped"}
-
-        api = package.api
-        monitor = package._global_image_monitor
-        real_stop = monitor.stop_monitoring
-        real_close = api.close_auto_thumbnails
-
-        def stop_then_advance(timeout):
-            ok = real_stop(timeout)
-            held.now[0] += advance
-            return ok
-
-        def recorded_close(final=False, timeout=None):
-            held.timeouts.append(timeout)
-            return real_close(final=final, timeout=timeout)
-
-        for target, name, value in (
-            (api, "_get_all_output_dirs", lambda: [Path(previews)]),
-            (api, "_generate_one", held_encode),
-            (api, "close_auto_thumbnails", recorded_close),
-            (monitor, "stop_monitoring", stop_then_advance),
-            (package, "_clock", lambda: held.now[0]),
-        ):
-            patcher = mock.patch.object(target, name, value)
-            patcher.start()
-            self.addCleanup(patcher.stop)
-        self.assertEqual(api.queue_capture_thumbnail(image), "queued")
-        self.addCleanup(held.release.set)
-        self.assertTrue(held.entered.wait(WAIT_SECONDS), "the encode never started")
-        held.thumbnailer = api._auto_thumbnailer  # the teardown drops the API's
-        held.worker = held.thumbnailer._worker
-        # Cleanups run newest first: release the encode, then join the worker
-        # (even when an assertion failed first), then end the patches.
-        self.addCleanup(held.worker.join, WAIT_SECONDS)
-        self.addCleanup(held.release.set)
-        self.assertTrue(held.worker.is_alive())
-        return held
+        """The image worker alive in its processing delay."""
+        handler = package._global_image_monitor.handler
+        self.assertIsNotNone(handler, "the package's monitor did not start")
+        handler.processing_delay = 3600
+        # Outside the watched folder, so the observer does not schedule it first.
+        path = os.path.join(self.tmp.name, name)
+        with open(path, "wb") as image:
+            image.write(b"x" * 16)
+        self.assertTrue(handler.schedule(path))
+        self.assertTrue(handler._worker.is_alive())
+        return handler._worker
 
 
 class TestPackageTeardown(PackageTeardownCase):
+    def test_teardown_stops_every_part_once(self):
+        package = self.load()
+        monitor = package._global_image_monitor
+        observer = monitor.observer
+        tracker_thread = package._global_prompt_tracker.cleanup_thread
+        worker = self.busy_worker(package)
+
+        with self.assertLogs("prompt_manager.init", "INFO") as logs:
+            result = package._stop_background_work()
+
+        self.assertEqual(result, {"monitor": True, "thumbnails": True, "tracker": True})
+        self.assertEqual(list(result), ["monitor", "thumbnails", "tracker"])
+        self.assertIn("Background work stopped in", "\n".join(logs.output))
+        self.assertIn("monitor=ok thumbnails=ok tracker=ok", "\n".join(logs.output))
+        self.assertFalse(worker.is_alive())
+        self.assertFalse(observer.is_alive())
+        self.assertFalse(tracker_thread.is_alive())
+        result["monitor"] = False  # a copy: the stored result is unchanged
+        with mock.patch.object(monitor, "stop_monitoring") as again:
+            self.assertEqual(
+                package._stop_background_work(),
+                {"monitor": True, "thumbnails": True, "tracker": True},
+            )
+        again.assert_not_called()
+
+    def test_teardown_detaches_the_monitor_callbacks(self):
+        package = self.load()
+        monitor = package._global_image_monitor
+        self.assertIsNotNone(monitor.review_recorder)
+        package._stop_background_work()
+        self.assertIsNone(monitor.review_recorder)
+        self.assertIsNone(monitor.thumbnail_queue)
+        self.assertIsNone(monitor.output_listener)
+
     def test_an_image_in_hand_queues_no_preview_during_teardown(self):
         package = self.load()
         monitor = package._global_image_monitor
@@ -759,17 +712,11 @@ class TestPackageTeardown(PackageTeardownCase):
         self.assertIsNone(seen["worker"])
         self.assertTrue(all(result.values()), result)
 
-    def test_teardown_kills_ffmpeg_and_stops_the_preview_worker(self):
+    def test_teardown_removes_its_atexit_callback(self):
         package = self.load()
-        held = self.hold_a_preview(package, advance=4.0, kill_releases=True)
-
-        result = package._stop_background_work()
-
-        self.assertTrue(held.killed.is_set())
-        self.assertEqual(held.timeouts, [6.0])  # min(30, 10 - 4)
-        self.assertTrue(result["thumbnails"], result)
-        self.assertFalse(held.worker.is_alive())
-        self.assertEqual(held.stop_seen, [True])
+        with mock.patch("atexit.unregister", wraps=atexit.unregister) as unregister:
+            package._stop_background_work()
+        unregister.assert_called_once_with(package._stop_background_work)
 
     def test_on_cleanup_runs_the_teardown(self):
         app = hooked_app()
@@ -802,74 +749,6 @@ class TestPackageTeardown(PackageTeardownCase):
         )
 
 
-class ListHandler(logging.Handler):
-    def __init__(self):
-        super().__init__(logging.INFO)
-        self.lines = []
-
-    def emit(self, record):
-        self.lines.append(f"{record.levelname}:{record.getMessage()}")
-
-
-class TestPackageTeardownOnce(PackageTestCase):
-    """The tests that only observe one plain teardown share one package.
-
-    The teardown runs once, so it runs here in ``setUpClass`` (per-test
-    ``assertLogs`` cannot wrap it); the tests read what it recorded.
-    """
-
-    package_name = "pm_teardown_once"
-
-    @classmethod
-    def setUpClass(cls):
-        with contextlib.redirect_stdout(io.StringIO()):
-            super().setUpClass()
-        package = cls.package
-        cls.monitor = package._global_image_monitor
-        cls.observer = cls.monitor.observer
-        cls.tracker_thread = package._global_prompt_tracker.cleanup_thread
-        cls.worker = start_busy_worker(package, cls.tmp.name)
-        cls.recorder_before = cls.monitor.review_recorder
-        logger = logging.getLogger("prompt_manager.init")
-        handler, level = ListHandler(), logger.level
-        logger.addHandler(handler)
-        logger.setLevel(logging.INFO)  # otherwise set by the config
-        try:
-            with mock.patch("atexit.unregister", wraps=atexit.unregister) as unregister:
-                cls.result = package._stop_background_work()
-        finally:
-            logger.removeHandler(handler)
-            logger.setLevel(level)
-        cls.log = "\n".join(handler.lines)
-        cls.unregister_calls = unregister.call_args_list
-
-    def test_teardown_stops_every_part_once(self):
-        expected = {"monitor": True, "thumbnails": True, "tracker": True}
-        self.assertEqual(self.result, expected)
-        self.assertEqual(list(self.result), ["monitor", "thumbnails", "tracker"])
-        self.assertIn("Background work stopped in", self.log)
-        self.assertIn("monitor=ok thumbnails=ok tracker=ok", self.log)
-        self.assertFalse(self.worker.is_alive())
-        self.assertFalse(self.observer.is_alive())
-        self.assertFalse(self.tracker_thread.is_alive())
-        copy = self.package._stop_background_work()
-        copy["monitor"] = False  # a copy: the stored result is unchanged
-        with mock.patch.object(self.monitor, "stop_monitoring") as again:
-            self.assertEqual(self.package._stop_background_work(), expected)
-        again.assert_not_called()
-
-    def test_teardown_detaches_the_monitor_callbacks(self):
-        self.assertIsNotNone(self.recorder_before)
-        self.assertIsNone(self.monitor.review_recorder)
-        self.assertIsNone(self.monitor.thumbnail_queue)
-        self.assertIsNone(self.monitor.output_listener)
-
-    def test_teardown_removes_its_atexit_callback(self):
-        self.assertEqual(
-            self.unregister_calls, [mock.call(self.package._stop_background_work)]
-        )
-
-
 class TestPackageTeardownFailures(PackageTeardownCase):
     """A failing step never skips the later ones (design D14, R2)."""
 
@@ -896,50 +775,6 @@ class TestPackageTeardownFailures(PackageTeardownCase):
     def test_second_interrupt_does_not_skip_the_tracker(self):
         self.assert_tracker_stops_after(KeyboardInterrupt())
 
-    def test_an_interrupt_between_steps_still_records_the_teardown(self):
-        package = self.load(require_stopped=False)
-        real_step = package._teardown_step
-
-        def interrupted_before_the_previews(name, stop, logger):
-            if name == "the preview worker":
-                raise KeyboardInterrupt  # outside the step's own try
-            return real_step(name, stop, logger)
-
-        with (
-            mock.patch.object(
-                package, "_teardown_step", interrupted_before_the_previews
-            ),
-            mock.patch("atexit.unregister", wraps=atexit.unregister) as unregister,
-            self.assertRaises(KeyboardInterrupt),
-        ):
-            package._stop_background_work()
-        expected = {"monitor": True, "thumbnails": False, "tracker": False}
-        self.assertEqual(package._background_work_result, expected)
-        unregister.assert_called_once_with(package._stop_background_work)
-        with mock.patch.object(
-            package._global_image_monitor, "stop_monitoring"
-        ) as again:
-            self.assertEqual(package._stop_background_work(), expected)
-        again.assert_not_called()
-
-    def test_teardown_reports_a_preview_worker_still_held(self):
-        package = self.load(require_stopped=False)
-        held = self.hold_a_preview(package, advance=9.95, kill_releases=False)
-
-        with self.assertLogs("prompt_manager.init", "WARNING") as logs:
-            result = package._stop_background_work()
-
-        self.assertTrue(held.killed.is_set())
-        self.assertEqual(len(held.timeouts), 1)
-        self.assertAlmostEqual(held.timeouts[0], 0.05, places=6)
-        self.assertTrue(held.thumbnailer.closed)
-        self.assertFalse(result["thumbnails"], result)
-        self.assertRegex("\n".join(logs.output), r"not stopped in time .*thumbnails")
-        held.release.set()
-        held.worker.join(WAIT_SECONDS)
-        self.assertFalse(held.worker.is_alive())
-        self.assertEqual(held.stop_seen, [True])
-
     def test_teardown_shares_one_deadline(self):
         package = self.load(require_stopped=False)
         now = [100.0]

tests/test_lora_import.py

@@ -529,16 +529,6 @@ class LoraImportRouteTests(LoraImportAPITestCase):
         resp, body = await self._post(APPLY, {"token": plan["token"]})
         self.assertEqual((resp.status, body["code"]), (409, "plan-used"))
 
-    async def test_a_finished_import_clears_the_layout_cache(self):
-        # #362: the import may link files under a folder LoraManager created
-        # after the image routes cached its layout.
-        self._two_loras()
-        with patch("py.lora_utils.clear_layout_cache") as clear:
-            _plan, job = await self._apply_reviewed()
-            await job.task
-        self.assertEqual(job.state, "finished", job.log)
-        clear.assert_called_once_with()
-
     async def test_discovery_failure_keeps_previous_imports(self):
         self._two_loras()
         before = _snapshot(self.api.db)
@@ -946,35 +936,6 @@ class LoraImportRefusalTests(LoraImportAPITestCase):
         self.assertEqual((job.state, job.scope["failure"]), ("failed", "plan-changed"))
         self.assertEqual(_snapshot(self.api.db), before)
 
-    async def test_a_refused_import_keeps_the_layout_cache(self):
-        self._two_loras()
-        resp, plan = await self._post(PREVIEW)
-        self.api.db.save_prompt(
-            text="late", category=CATEGORY, prompt_hash=generate_prompt_hash("late")
-        )
-        with patch("py.lora_utils.clear_layout_cache") as clear:
-            resp, body = await self._post(APPLY, {"token": plan["token"]})
-            job = self.api._jobs.get(body["job"]["id"])
-            await job.task
-        self.assertEqual(job.scope["failure"], "plan-changed")
-        clear.assert_not_called()
-
-    async def test_start_refused_when_a_skipped_sidecar_changed(self):
-        # #362 CUJ 4: the review counted the orphan as skipped.
-        self._two_loras()
-        orphan = self.loras / "orphan.metadata.json"
-        orphan.write_text(json.dumps({"model_name": "orphan"}))
-        resp, plan = await self._post(PREVIEW)
-        self.assertEqual(resp.status, 200, plan)
-        self.assertEqual((plan["lora_files"], plan["skipped"]), (3, 1))
-        before = _snapshot(self.api.db)
-        orphan.write_text(orphan.read_text() + "  ")
-        resp, body = await self._post(APPLY, {"token": plan["token"]})
-        self.assertEqual(resp.status, 409, body)
-        self.assertEqual(body["code"], "plan-changed")
-        self.assertEqual(self.api._jobs.list(), [])
-        self.assertEqual(_snapshot(self.api.db), before)
-
     async def test_a_pair_saved_after_the_review_blocks_the_job(self):
         self._two_loras()
         resp, plan = await self._post(PREVIEW)
@@ -1226,62 +1187,6 @@ class LoraImportStageTests(unittest.TestCase):
         )
         self.assertNotEqual(lora_import.fingerprint(self.discover()), first)
 
-    # #362: a skipped sidecar is part of what the review counted.
-
-    def _assert_fingerprint_follows(self, change):
-        first = lora_import.fingerprint(self.discover())
-        self.assertEqual(lora_import.fingerprint(self.discover()), first)
-        change()
-        self.assertNotEqual(lora_import.fingerprint(self.discover()), first)
-
-    def _assert_add_edit_remove_change_it(self, path, text):
-        self._assert_fingerprint_follows(lambda: path.write_text(text))
-        self._assert_fingerprint_follows(
-            lambda: path.write_text(path.read_text() + "  ")
-        )
-        self._assert_fingerprint_follows(path.unlink)
-
-    def _centralized(self):
-        sidecars = self.tmp / "sidecars"
-        sidecars.mkdir()
-        self.settings_file(
-            {
-                "sidecar_storage_mode": "centralized",
-                "sidecar_storage_path": str(sidecars),
-            }
-        )
-        model = self.loras / "kept.safetensors"
-        model.write_bytes(b"m")
-        (sidecars / "kept.metadata.json").write_text(
-            json.dumps({"file_path": str(model)})
-        )
-        return sidecars
-
-    def test_fingerprint_follows_a_centralized_sidecar_whose_model_is_missing(self):
-        sidecars = self._centralized()
-        gone = self.loras / "gone.safetensors"
-        self._assert_add_edit_remove_change_it(
-            sidecars / "gone.metadata.json", json.dumps({"file_path": str(gone)})
-        )
-        self.assertEqual(len(self.discover().sidecars), 1)
-
-    def test_fingerprint_follows_a_checkpoint_sidecar_outside_the_lora_roots(self):
-        sidecars = self._centralized()
-        checkpoints = self.tmp / "checkpoints"
-        checkpoints.mkdir()
-        ckpt = checkpoints / "base.safetensors"
-        ckpt.write_bytes(b"c")
-        self._assert_add_edit_remove_change_it(
-            sidecars / "base.metadata.json", json.dumps({"file_path": str(ckpt)})
-        )
-
-    def test_fingerprint_follows_an_alongside_sidecar_without_its_model(self):
-        _write_metadata(self.loras, "kept")
-        self._assert_add_edit_remove_change_it(
-            self.loras / "orphan.metadata.json", json.dumps({"model_name": "orphan"})
-        )
-        self.assertEqual(len(self.discover().sidecars), 1)
-
     def test_fingerprint_follows_a_missing_file_between_stat_calls(self):
         meta = _write_metadata(self.loras, "one")
         discovery = self.discover()

tests/test_lora_layout_cache.py

@@ -1,590 +0,0 @@
-"""The LoraManager layout cache for the image hot paths (#362).
-
-``py/lora_layout_cache.py``. Every test injects the install resolver, the
-layout loader (both counting calls), the wall clock, the monotonic clock and,
-where needed, ``stat`` and ``pathmod``, over a temporary folder. No test
-sleeps: modification times are set with ``os.utime`` and the clocks are
-plain attributes the test moves.
-"""
-
-import ntpath
-import os
-import shutil
-import sys
-import tempfile
-import threading
-import time
-import unittest
-from dataclasses import replace
-from pathlib import Path, PureWindowsPath
-from types import SimpleNamespace
-from unittest.mock import patch
-
-sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
-
-from py import lora_layout_cache as cache_module
-from py.lora_layout_cache import (
-    MAX_AGE_SECONDS,
-    MISSING,
-    RACY_SECONDS,
-    LayoutCache,
-)
-from py.lora_manager_layout import UNREADABLE, LoraManagerLayout, _canonical
-
-SECOND_NS = 10**9
-# NTFS keeps modification times in 100 ns steps and FAT in 2 s steps: every
-# time a test sets is a whole millisecond, and a test whose file system rounds
-# it anyway skips (``set_mtime``).
-TICK_NS = 10**6
-
-
-def make_layout(settings_path=None, media_roots=(), error=None, source="user config"):
-    return LoraManagerLayout(
-        settings_path=settings_path,
-        source=source,
-        error=error,
-        mode="alongside",
-        lora_roots=tuple(media_roots),
-        sidecar_root=None,
-        example_images_root=None,
-        library="default",
-        media_roots=tuple(media_roots),
-    )
-
-
-class CacheTestCase(unittest.TestCase):
-    """An install, its plugin settings file, a settings file and a LoRA root."""
-
-    def setUp(self):
-        self.tmp = Path(tempfile.mkdtemp()).resolve()
-        self.addCleanup(shutil.rmtree, self.tmp)
-        self.install_dir = self.tmp / "custom_nodes" / "ComfyUI-Lora-Manager"
-        self.install_dir.mkdir(parents=True)
-        self.plugin_settings = self.install_dir / "settings.json"
-        self.settings = self.tmp / "user-config" / "settings.json"
-        self.settings.parent.mkdir()
-        self.root_dir = self.tmp / "loras"
-        self.root_dir.mkdir()
-        # Files are written ten minutes before the injected wall clock, so
-        # nothing is racy unless a test sets a recent modification time.
-        self.now_ns = time.time_ns() // TICK_NS * TICK_NS
-        self.mono = 1000.0
-        self.resolved = []
-        self.loaded = []
-        self.install = _canonical(str(self.install_dir))
-        self.layout = make_layout(
-            settings_path=self.settings,
-            media_roots=(_canonical(str(self.root_dir)),),
-        )
-        self.write(self.settings, "{}")
-
-    def write(self, path, text, mtime_ns=None):
-        Path(path).write_text(text, encoding="utf-8")
-        stamp = mtime_ns if mtime_ns is not None else self.now_ns - 600 * SECOND_NS
-        self.set_mtime(path, stamp)
-
-    def set_mtime(self, path, stamp):
-        os.utime(path, ns=(stamp, stamp))
-        if os.stat(path).st_mtime_ns != stamp:
-            self.skipTest("this file system rounds modification times")
-
-    def resolve(self, key):
-        self.resolved.append(key)
-        return self.install
-
-    def load(self, install):
-        self.loaded.append(install)
-        return self.layout
-
-    def cache(self, **kwargs):
-        kwargs.setdefault("wall_ns", lambda: self.now_ns)
-        kwargs.setdefault("monotonic", lambda: self.mono)
-        return LayoutCache(
-            kwargs.pop("resolve", self.resolve), kwargs.pop("load", self.load), **kwargs
-        )
-
-
-class ValidationTests(CacheTestCase):
-    def test_hit_reuses_the_entry_without_resolving_or_loading(self):
-        cache = self.cache()
-        first = cache.get("")
-        second = cache.get("")
-        self.assertEqual(len(self.resolved), 1)
-        self.assertEqual(len(self.loaded), 1)
-        self.assertIs(first[1], second[1])
-        self.assertIs(first[0], second[0])
-        self.assertEqual(first, (self.install, self.layout))
-
-    def test_settings_size_change_reloads_once(self):
-        cache = self.cache()
-        cache.get("")
-        stamp = self.settings.stat().st_mtime_ns
-        self.write(self.settings, '{"a": 1}', mtime_ns=stamp)
-        cache.get("")
-        cache.get("")
-        self.assertEqual(len(self.loaded), 2)
-
-    def test_settings_mtime_change_reloads_once(self):
-        cache = self.cache()
-        cache.get("")
-        self.set_mtime(self.settings, self.settings.stat().st_mtime_ns + TICK_NS)
-        cache.get("")
-        cache.get("")
-        self.assertEqual(len(self.loaded), 2)
-
-    def test_settings_deletion_reloads_once(self):
-        cache = self.cache()
-        cache.get("")
-        self.settings.unlink()
-        cache.get("")
-        cache.get("")
-        self.assertEqual(len(self.loaded), 2)
-
-    def test_settings_creation_reloads_once(self):
-        self.settings.unlink()
-        cache = self.cache()
-        cache.get("")
-        self.write(self.settings, "{}")
-        cache.get("")
-        cache.get("")
-        self.assertEqual(len(self.loaded), 2)
-
-    def test_plugin_settings_change_reloads_once(self):
-        cache = self.cache()
-        cache.get("")
-        self.write(self.plugin_settings, '{"use_portable_settings": true}')
-        cache.get("")
-        cache.get("")
-        self.assertEqual(len(self.loaded), 2)
-
-    def test_a_different_configured_path_reloads(self):
-        cache = self.cache()
-        cache.get("")
-        cache.get("other")
-        self.assertEqual(self.resolved, ["", "other"])
-        self.assertEqual(len(self.loaded), 2)
-
-    def test_a_different_home_folder_reloads(self):
-        # The folder guard compares against the home folder: a cached layout
-        # must not outlive a change of it.
-        home = {"value": str(self.tmp / "home-a")}
-        cache = self.cache(home=lambda: home["value"])
-        cache.get("")
-        cache.get("")
-        home["value"] = str(self.tmp / "home-b")
-        cache.get("")
-        self.assertEqual(len(self.loaded), 2)
-
-    def test_configured_path_is_stripped_for_the_key_and_the_resolver(self):
-        cache = self.cache()
-        cache.get("  x ")
-        cache.get("x")
-        self.assertEqual(self.resolved, ["x"])
-        self.assertEqual(len(self.loaded), 1)
-
-    def test_a_recently_modified_file_is_not_stored(self):
-        self.write(self.settings, "{}", mtime_ns=self.now_ns - SECOND_NS)
-        cache = self.cache()
-        cache.get("")
-        cache.get("")
-        self.assertEqual(len(self.loaded), 2)
-        self.assertIsNone(cache._entry)
-        self.now_ns += RACY_SECONDS * SECOND_NS
-        cache.get("")
-        self.assertIsNotNone(cache._entry)
-        cache.get("")
-        self.assertEqual(len(self.loaded), 3)
-
-    def test_racy_boundary(self):
-        # Exactly RACY_SECONDS old is trusted; anything newer is not.
-        boundary = self.now_ns - RACY_SECONDS * SECOND_NS
-        self.write(self.settings, "{}", mtime_ns=boundary + TICK_NS)
-        cache = self.cache()
-        cache.get("")
-        self.assertIsNone(cache._entry)
-        self.set_mtime(self.settings, boundary)
-        cache.get("")
-        self.assertIsNotNone(cache._entry)
-
-    def test_a_modification_time_in_the_future_is_racy(self):
-        self.write(self.plugin_settings, "{}", mtime_ns=self.now_ns + 60 * SECOND_NS)
-        cache = self.cache()
-        cache.get("")
-        cache.get("")
-        self.assertEqual(len(self.loaded), 2)
-        self.assertIsNone(cache._entry)
-
-    def test_a_missing_file_is_not_racy(self):
-        self.assertFalse(self.plugin_settings.exists())
-        self.settings.unlink()
-        cache = self.cache()
-        cache.get("")
-        self.assertIsNotNone(cache._entry)
-        signatures = dict(cache._entry.signatures)
-        self.assertTrue(all(value is MISSING for value in signatures.values()))
-        self.assertEqual(len(signatures), 2)
-
-    def test_entry_is_reused_below_the_maximum_age_and_reloaded_at_it(self):
-        cache = self.cache()
-        cache.get("")
-        stored_at = self.mono
-        self.mono = stored_at + MAX_AGE_SECONDS - 0.5
-        cache.get("")
-        self.assertEqual(len(self.loaded), 1)
-        self.mono = stored_at + MAX_AGE_SECONDS
-        cache.get("")
-        self.assertEqual(len(self.loaded), 2)
-
-    def test_a_wall_clock_moved_backward_does_not_extend_the_age(self):
-        cache = self.cache()
-        cache.get("")
-        self.now_ns -= 3600 * SECOND_NS
-        self.mono += MAX_AGE_SECONDS
-        # The old entry is too old; the new load sees files "from the future"
-        # (racy) and is not stored, but it does load.
-        cache.get("")
-        self.assertEqual(len(self.loaded), 2)
-
-    def test_stat_failure_on_a_hit_reloads(self):
-        failing = {"on": False}
-
-        def stat(path):
-            if failing["on"]:
-                raise PermissionError(13, "sharing violation", path)
-            return os.stat(path)
-
-        cache = self.cache(stat=stat)
-        cache.get("")
-        failing["on"] = True
-        cache.get("")
-        self.assertEqual(len(self.loaded), 2)
-        self.assertIsNone(cache._entry)
-
-    def test_stat_failure_on_a_miss_leaves_the_layout_unstored(self):
-        def stat(path):
-            raise PermissionError(13, "sharing violation", path)
-
-        cache = self.cache(stat=stat)
-        self.assertEqual(cache.get(""), (self.install, self.layout))
-        self.assertIsNone(cache._entry)
-        cache.get("")
-        self.assertEqual(len(self.loaded), 2)
-
-    def test_an_unreadable_settings_layout_is_not_stored(self):
-        self.layout = make_layout(settings_path=self.settings, error=UNREADABLE)
-        cache = self.cache()
-        self.assertEqual(cache.get("")[1].error, UNREADABLE)
-        cache.get("")
-        self.assertEqual(len(self.loaded), 2)
-        self.assertIsNone(cache._entry)
-
-    def test_other_layout_errors_are_stored(self):
-        self.layout = make_layout(
-            settings_path=self.settings, error="the settings file is not valid JSON"
-        )
-        cache = self.cache()
-        cache.get("")
-        cache.get("")
-        self.assertEqual(len(self.loaded), 1)
-
-    def _symlink(self, link, target):
-        try:
-            os.symlink(target, link, target_is_directory=True)
-        except (OSError, NotImplementedError) as error:
-            self.skipTest(f"symlinks are not available here: {error}")
-
-    def test_a_media_root_swapped_for_a_symlink_reloads(self):
-        elsewhere = self.tmp / "elsewhere"
-        elsewhere.mkdir()
-        cache = self.cache()
-        cache.get("")
-        shutil.rmtree(self.root_dir)
-        self._symlink(self.root_dir, elsewhere)
-        cache.get("")
-        self.assertEqual(len(self.loaded), 2)
-
-    def test_an_install_swapped_for_a_symlink_reloads(self):
-        elsewhere = self.tmp / "elsewhere-lm"
-        elsewhere.mkdir()
-        cache = self.cache()
-        cache.get("")
-        moved = self.tmp / "moved-lm"
-        self.install_dir.rename(moved)
-        self._symlink(self.install_dir, elsewhere)
-        cache.get("")
-        self.assertEqual(len(self.loaded), 2)
-
-    def test_a_root_whose_realpath_fails_reloads(self):
-        cache = self.cache()
-        cache.get("")
-        with patch.object(
-            cache_module, "_canonical", side_effect=OSError("realpath failed")
-        ):
-            cache.get("")
-        self.assertEqual(len(self.loaded), 2)
-
-    def test_a_stat_value_error_cannot_validate(self):
-        def stat(path):
-            raise ValueError("embedded null character in path")
-
-        cache = self.cache(stat=stat)
-        cache.get("")
-        self.assertIsNone(cache._entry)
-
-    def test_not_a_directory_is_missing(self):
-        def stat(path):
-            raise NotADirectoryError(20, "Not a directory", path)
-
-        cache = self.cache(stat=stat)
-        cache.get("")
-        self.assertTrue(
-            all(value is MISSING for _path, value in cache._entry.signatures)
-        )
-        cache.get("")
-        self.assertEqual(len(self.loaded), 1)
-
-    def test_an_offline_windows_share_cannot_validate(self):
-        def stat(path):
-            error = FileNotFoundError(2, "The network path was not found", path)
-            error.winerror = 53
-            raise error
-
-        cache = self.cache(stat=stat)
-        cache.get("")
-        self.assertIsNone(cache._entry)
-
-    def test_a_failed_reload_leaves_no_stale_entry(self):
-        failing = {"on": False}
-
-        def load(install):
-            self.loaded.append(install)
-            if failing["on"]:
-                raise RuntimeError("boom")
-            return self.layout
-
-        cache = self.cache(load=load)
-        cache.get("")
-        self.mono += MAX_AGE_SECONDS
-        failing["on"] = True
-        with self.assertRaises(RuntimeError):
-            cache.get("")
-        self.assertIsNone(cache._entry)
-
-    def test_a_resolve_failure_propagates_and_stores_nothing(self):
-        def resolve(key):
-            raise RuntimeError("boom")
-
-        cache = self.cache(resolve=resolve)
-        with self.assertRaises(RuntimeError):
-            cache.get("")
-        self.assertIsNone(cache._entry)
-
-    def test_without_a_settings_path_only_the_plugin_file_is_watched(self):
-        self.layout = make_layout(media_roots=self.layout.media_roots)
-        cache = self.cache()
-        cache.get("")
-        self.assertEqual(
-            [path for path, _sig in cache._entry.signatures],
-            [os.path.join(self.install, "settings.json")],
-        )
-
-    def test_portable_settings_are_watched_once(self):
-        self.layout = make_layout(settings_path=self.install_dir / "settings.json")
-        cache = self.cache()
-        cache.get("")
-        self.assertEqual(len(cache._entry.signatures), 1)
-
-    def test_a_load_failure_propagates_and_stores_nothing(self):
-        calls = []
-
-        def load(install):
-            calls.append(install)
-            if len(calls) == 1:
-                raise RuntimeError("boom")
-            return self.layout
-
-        cache = self.cache(load=load)
-        with self.assertRaises(RuntimeError):
-            cache.get("")
-        self.assertIsNone(cache._entry)
-        self.assertEqual(cache.get(""), (self.install, self.layout))
-        self.assertEqual(len(calls), 2)
-
-    def test_without_an_install_nothing_is_watched_and_the_entry_ages_out(self):
-        stats = []
-
-        def stat(path):
-            stats.append(path)
-            return os.stat(path)
-
-        self.layout = make_layout(media_roots=self.layout.media_roots, source=None)
-        cache = self.cache(resolve=lambda key: self.resolved.append(key), stat=stat)
-        self.assertEqual(cache.get(""), (None, self.layout))
-        self.assertEqual(cache._entry.signatures, ())
-        cache.get("")
-        self.assertEqual(len(self.loaded), 1)
-        self.assertEqual(stats, [])
-        self.mono += MAX_AGE_SECONDS
-        cache.get("")
-        self.assertEqual(len(self.loaded), 2)
-
-    def test_windows_paths_differing_only_in_case_are_watched_once(self):
-        stats = []
-
-        def stat(path):
-            stats.append(path)
-            return SimpleNamespace(st_mtime_ns=self.now_ns - 600 * SECOND_NS, st_size=2)
-
-        install = _canonical("C:\\ComfyUI\\custom_nodes\\LM", ntpath)
-        # The hit check needs ntpath's realpath to be idempotent here.
-        self.assertEqual(install, _canonical(install, ntpath))
-        self.layout = make_layout(
-            settings_path=PureWindowsPath(
-                "C:\\ComfyUI\\custom_nodes\\lm\\SETTINGS.JSON"
-            )
-        )
-        cache = self.cache(resolve=lambda key: install, stat=stat, pathmod=ntpath)
-        cache.get("")
-        self.assertEqual(len(cache._entry.signatures), 1)
-        self.assertEqual(len(stats), 1)
-        cache.get("")
-        self.assertEqual(len(self.loaded), 1)
-        self.assertEqual(len(stats), 2)
-
-
-class ConcurrencyTests(CacheTestCase):
-    def test_a_load_that_started_before_clear_stores_nothing(self):
-        started = threading.Event()
-        release = threading.Event()
-
-        def load(install):
-            self.loaded.append(install)
-            if len(self.loaded) == 1:
-                started.set()
-                self.assertTrue(release.wait(10))
-            return self.layout
-
-        cache = self.cache(load=load)
-        results = []
-        worker = threading.Thread(target=lambda: results.append(cache.get("")))
-        worker.start()
-        self.assertTrue(started.wait(10))
-        cache.clear()
-        release.set()
-        worker.join(10)
-        self.assertFalse(worker.is_alive())
-        self.assertEqual(results, [(self.install, self.layout)])
-        self.assertIsNone(cache._entry)
-        cache.get("")
-        self.assertEqual(len(self.loaded), 2)
-        self.assertIsNotNone(cache._entry)
-
-    def test_concurrent_misses_all_get_a_layout_and_one_entry_is_stored(self):
-        cache = self.cache()
-        count = 6
-        barrier = threading.Barrier(count)
-        results = []
-        lock = threading.Lock()
-
-        def worker():
-            barrier.wait(10)
-            result = cache.get("")
-            with lock:
-                results.append(result)
-
-        threads = [threading.Thread(target=worker) for _ in range(count)]
-        for thread in threads:
-            thread.start()
-        for thread in threads:
-            thread.join(10)
-        self.assertEqual(results, [(self.install, self.layout)] * count)
-        self.assertIsNotNone(cache._entry)
-        loads = len(self.loaded)
-        cache.get("")
-        self.assertEqual(len(self.loaded), loads)
-
-    def test_a_valid_hit_takes_no_lock(self):
-        cache = self.cache()
-        cache.get("")
-        results = []
-        with cache._lock:
-            worker = threading.Thread(target=lambda: results.append(cache.get("")))
-            worker.start()
-            worker.join(10)
-            alive = worker.is_alive()
-        worker.join(10)
-        self.assertFalse(alive)
-        self.assertEqual(results, [(self.install, self.layout)])
-
-    def test_clear_forces_a_reload(self):
-        cache = self.cache()
-        cache.get("")
-        cache.clear()
-        self.assertIsNone(cache._entry)
-        cache.get("")
-        self.assertEqual(len(self.loaded), 2)
-
-
-class LoggingTests(CacheTestCase):
-    LOGGER = "prompt_manager.lora_layout_cache"
-
-    def get(self, cache):
-        with self.assertLogs(self.LOGGER, "DEBUG") as logs:
-            cache.get("")
-            # assertLogs needs one record; this marker is not counted.
-            cache_module.logger.debug("marker")
-        return [r for r in logs.records if r.getMessage() != "marker"]
-
-    def test_a_new_layout_logs_one_info_line(self):
-        cache = self.cache()
-        records = self.get(cache)
-        self.assertEqual([r.levelname for r in records], ["INFO"])
-        self.assertEqual(
-            records[0].getMessage(),
-            "LoraManager layout changed: user config settings, alongside storage, "
-            "1 media folders",
-        )
-        self.mono += MAX_AGE_SECONDS
-        self.assertEqual(self.get(cache), [])  # an equal layout, reloaded
-        cache.clear()
-        self.assertEqual(self.get(cache), [])
-        self.layout = replace(self.layout, media_roots=(), lora_roots=())
-        self.mono += MAX_AGE_SECONDS
-        self.assertEqual(len(self.get(cache)), 1)
-
-    def test_no_settings_source_prints_no(self):
-        self.layout = replace(self.layout, source=None)
-        records = self.get(self.cache())
-        self.assertIn("layout changed: no settings,", records[0].getMessage())
-
-    def test_an_unstored_load_logs_no_info_line(self):
-        self.write(self.settings, "{}", mtime_ns=self.now_ns)
-        self.assertEqual(
-            [r for r in self.get(self.cache()) if r.levelname == "INFO"], []
-        )
-
-    def test_a_newly_refused_default_folder_logs_one_warning(self):
-        self.layout = replace(self.layout, refused_defaults=("/home",))
-        cache = self.cache()
-        records = self.get(cache)
-        warnings = [r for r in records if r.levelname == "WARNING"]
-        self.assertEqual(len(warnings), 1)
-        self.assertIn("/home", warnings[0].getMessage())
-        self.mono += MAX_AGE_SECONDS
-        self.assertEqual(self.get(cache), [])
-        cache.clear()
-        self.assertEqual(self.get(cache), [])
-        self.layout = replace(self.layout, refused_defaults=("/home", "/"))
-        self.mono += MAX_AGE_SECONDS
-        warnings = [r for r in self.get(cache) if r.levelname == "WARNING"]
-        self.assertEqual([w.getMessage().count("'/'") for w in warnings], [1])
-
-
-class ModuleTests(unittest.TestCase):
-    def test_constants(self):
-        self.assertEqual(cache_module.RACY_SECONDS, 3)
-        self.assertEqual(cache_module.MAX_AGE_SECONDS, 30)
-
-
-if __name__ == "__main__":
-    unittest.main()

tests/test_lora_manager_layout.py

@@ -33,10 +33,6 @@ MEDIA = frozenset(
     {".png", ".jpg", ".jpeg", ".webp", ".gif", ".mp4", ".webm", ".mov", ".bmp"}
 )
 MARKER = "marker-from-unkept-settings-0123"
-BAD_SIDECAR_REASON = (
-    "the sidecar folder is a filesystem root, the home folder or a folder "
-    "that contains it"
-)
 
 
 class LayoutTestCase(unittest.TestCase):
@@ -265,25 +261,10 @@ class UnusableLayoutTests(LayoutTestCase):
             {"sidecar_storage_mode": "centralized", "sidecar_storage_path": "/"},
         )
         layout = self.load()
-        self.assertEqual(layout.error, BAD_SIDECAR_REASON)
-        self.assertNotIn(os.path.normcase(os.path.realpath("/")), layout.media_roots)
-
-    def test_a_centralized_sidecar_root_that_contains_home(self):
-        home = self.tmp / "outer" / "me"
-        home.mkdir(parents=True)
-        self.write_settings(
-            self.user_config,
-            {
-                "sidecar_storage_mode": "centralized",
-                "sidecar_storage_path": str(self.tmp / "outer"),
-            },
-        )
-        with patch.object(layout_module.Path, "home", return_value=home):
-            layout = self.load()
-        self.assertEqual(layout.error, BAD_SIDECAR_REASON)
         self.assertEqual(
-            layout.media_roots, (os.path.normcase(str(self.default_loras)),)
+            layout.error, "the sidecar folder is a filesystem root or the home folder"
         )
+        self.assertNotIn(os.path.normcase(os.path.realpath("/")), layout.media_roots)
 
 
 class SecretTests(LayoutTestCase):
@@ -520,31 +501,6 @@ class CentralizedDiscoveryTests(LayoutTestCase):
         self.assertEqual(discovery.scanned, 3)
         self.assertEqual(discovery.sidecars, [])
 
-    def test_every_scanned_file_is_listed_resolved_and_sorted(self):
-        # #362: the fingerprint covers every scanned sidecar, kept or skipped.
-        ckpt = self.model(self.checkpoints, "base")
-        alpha = self.model(self.loras, "alpha")
-        self.sidecar(self.sidecars / "z", "base", {"file_path": str(ckpt)})
-        self.sidecar(self.sidecars / "a", "alpha", {"file_path": str(alpha)})
-        self.sidecar(self.sidecars / "m", "nopath", {"model_name": "x"})
-        discovery = discover_sidecars(self.load())
-        files = discovery.scanned_files
-        self.assertIsInstance(files, tuple)
-        self.assertEqual(len(files), 3)
-        self.assertEqual(discovery.scanned, len(files))
-        self.assertEqual([str(f) for f in files], sorted(str(f) for f in files))
-        self.assertEqual(
-            [f.name for f in files],
-            [
-                f"alpha{METADATA_SUFFIX}",
-                f"nopath{METADATA_SUFFIX}",
-                f"base{METADATA_SUFFIX}",
-            ],
-        )
-        for path in files:
-            self.assertEqual(str(path), os.path.realpath(path))
-        self.assertEqual(len(discovery.sidecars), 1)
-
     def test_leftover_sidecars_beside_models_are_not_scanned(self):
         model = self.model(self.loras, "alpha")
         self.sidecar(self.loras, "alpha", {"file_path": str(model)})
@@ -655,16 +611,6 @@ class LocalMediaTests(LayoutTestCase):
         self.touch(self.default_loras / "alpha.webp")
         self.assertEqual(self.media({"preview_url": preview}), [preview])
 
-    def test_a_relative_preview_url_is_logged_and_skipped(self):
-        fallback = self.touch(self.default_loras / "alpha.webp")
-        self.touch(self.default_loras / "relative.png")
-        with self.assertLogs("prompt_manager.lora_manager_layout", "DEBUG") as logs:
-            media = self.media({"preview_url": "relative.png"})
-        self.assertEqual(media, [fallback])
-        self.assertEqual(len(logs.records), 1)
-        self.assertEqual(logs.records[0].levelname, "DEBUG")
-        self.assertIn(self.meta.name, logs.output[0])
-
     def test_preview_url_outside_media_roots_or_not_servable_is_ignored(self):
         outside = self.touch(self.tmp / "elsewhere" / "p.png")
         avif = self.touch(self.default_loras / "p.avif")
@@ -864,63 +810,6 @@ class FolderGuardTests(unittest.TestCase):
         self.assertFalse(layout_module._inside(str(self.tmp / "other.png"), (root,)))
 
 
-class RefusedDefaultsTests(LayoutTestCase):
-    """#362: ComfyUI's default LoRA folders pass the same guard; the refused
-    ones that exist are listed so the layout cache can warn about them."""
-
-    def setUp(self):
-        super().setUp()
-        self.outer = self.tmp / "outer"
-        self.home = self.outer / "me"
-        self.home.mkdir(parents=True)
-        self.fs_root = os.path.abspath(os.sep)
-        self.defaults = [
-            str(self.home),
-            str(self.outer),
-            self.fs_root,
-            str(self.tmp / "missing"),
-            str(self.default_loras),
-        ]
-        self.expected = (
-            layout_module._canonical(str(self.home)),
-            layout_module._canonical(str(self.outer)),
-            layout_module._canonical(self.fs_root),
-        )
-        home_patch = patch.object(layout_module.Path, "home", return_value=self.home)
-        home_patch.start()
-        self.addCleanup(home_patch.stop)
-
-    def load_defaults(self, path="__plugin__"):
-        return self.load(path, default_roots=lambda: list(self.defaults))
-
-    def test_refused_existing_defaults_are_listed_on_a_usable_layout(self):
-        self.write_settings(self.user_config, {})
-        layout = self.load_defaults()
-        self.assertIsNone(layout.error)
-        self.assertEqual(layout.refused_defaults, self.expected)
-        self.assertEqual(
-            layout.lora_roots, (os.path.normcase(str(self.default_loras)),)
-        )
-
-    def test_they_are_listed_on_an_unusable_centralized_layout(self):
-        self.write_settings(
-            self.user_config,
-            {"sidecar_storage_mode": "centralized", "sidecar_storage_path": "/"},
-        )
-        layout = self.load_defaults()
-        self.assertEqual(layout.error, BAD_SIDECAR_REASON)
-        self.assertEqual(layout.refused_defaults, self.expected)
-
-    def test_they_are_listed_without_an_install(self):
-        layout = self.load_defaults(None)
-        self.assertEqual(layout.error, "no LoraManager install found")
-        self.assertEqual(layout.refused_defaults, self.expected)
-
-    def test_nothing_is_listed_when_every_default_is_accepted(self):
-        self.write_settings(self.user_config, {})
-        self.assertEqual(self.load().refused_defaults, ())
-
-
 class SummaryTests(LayoutTestCase):
 
     def test_summary_names_what_the_import_read(self):

tests/test_lora_utils.py

@@ -224,14 +224,6 @@ class TestReadLoraMetadata(unittest.TestCase):
         result = read_lora_metadata(Path("/nonexistent/file.metadata.json"))
         self.assertIsNone(result)
 
-    def test_returns_none_for_a_file_that_is_not_utf8(self):
-        # #362: one malformed sidecar is skipped; it must not fail the whole
-        # trigger-word load (now logged with a traceback) or the review.
-        with tempfile.TemporaryDirectory() as folder:
-            path = Path(folder) / "bad.metadata.json"
-            path.write_bytes(b'{"model_name": "\xff\xfe"}')
-            self.assertIsNone(read_lora_metadata(path))
-
 
 class TestGetLoraImageCacheDir(unittest.TestCase):
     """Test get_lora_image_cache_dir — returns and creates cache path."""
@@ -1799,34 +1791,6 @@ class TestTriggerWordCacheLayout(unittest.TestCase):
         self.assertTrue(cache.is_loaded)
         self.assertTrue(cache.retry_due())
 
-    def test_an_os_error_is_a_warning_without_a_traceback(self):
-        cache = TriggerWordCache()
-        cache._cache = {"old": ["word"]}
-        with (
-            patch.object(TriggerWordCache, "_build", side_effect=OSError("denied")),
-            self.assertLogs("prompt_manager.lora_utils", level="WARNING") as logs,
-        ):
-            self.assertEqual(cache.load(str(self.lm)), 1)
-        self.assertEqual([r.levelname for r in logs.records], ["WARNING"])
-        self.assertIsNone(logs.records[0].exc_info)
-        self.assertEqual(cache.get_trigger_words("old"), ["word"])
-        self.assertFalse(cache.retry_due())
-
-    def test_an_unexpected_failure_logs_its_traceback(self):
-        # #362: a programming error must not look like a disk error.
-        cache = TriggerWordCache()
-        cache._cache = {"old": ["word"]}
-        with (
-            patch.object(TriggerWordCache, "_build", side_effect=ValueError("bug")),
-            self.assertLogs("prompt_manager.lora_utils", level="WARNING") as logs,
-        ):
-            self.assertEqual(cache.load(str(self.lm)), 1)
-        self.assertEqual(len(logs.records), 1)
-        self.assertIsNotNone(logs.records[0].exc_info)
-        self.assertIs(logs.records[0].exc_info[0], ValueError)
-        self.assertEqual(cache.get_trigger_words("old"), ["word"])
-        self.assertFalse(cache.retry_due())
-
     def test_any_other_failure_never_raises(self):
         cache = TriggerWordCache()
         with (

tests/test_utils_logging_config.py

@@ -1,8 +1,10 @@
 """PromptManagerLogger: memory buffer, file rotation, level changes, truncation."""
 
 import logging
+import ntpath
 import os
 import pathlib
+import stat
 import sys
 import tempfile
 import threading
@@ -11,14 +13,20 @@ import unittest.mock as mock
 
 sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
 
-from utils import logging_config
-from utils.logging_config import (
+from utils import logging_config  # noqa: E402
+from utils.logging_config import (  # noqa: E402
     MemoryBufferHandler,
     PromptManagerLogger,
+    RotatedChanged,
     get_logger,
     get_logger_manager,
 )
 
+try:
+    from tests.open_handles import assert_closed  # noqa: E402
+except ImportError:  # discovered with tests/ as the top-level directory
+    from open_handles import assert_closed  # noqa: E402
+
 
 class IsolatedLoggerTestCase(unittest.TestCase):
     """A logger manager whose log directory is a temporary folder."""
@@ -32,6 +40,7 @@ class IsolatedLoggerTestCase(unittest.TestCase):
         with mock.patch.object(logging_config, "__file__", fake_file):
             self.manager = object.__new__(PromptManagerLogger)  # bypass singleton
             self.manager.__init__()
+        self.allowed_thread_names = set()
         self.manager.update_config({"console_logging": False})
         # Drop the "Updated logging configuration" entry.
         self.manager._log_buffer.clear()
@@ -53,12 +62,21 @@ class IsolatedLoggerTestCase(unittest.TestCase):
 
     def _filter_to_this_thread(self):
         this_thread = threading.get_ident()
+        # A test may let named helper threads of its own through too.
+        allowed_names = self.allowed_thread_names
+
+        def allowed(record):
+            return record.thread == this_thread or record.threadName in allowed_names
+
         for handler in self.manager.logger.handlers:
-            handler.addFilter(lambda record: record.thread == this_thread)
+            handler.filters.clear()
+            handler.addFilter(allowed)
 
     def _restore_global_logging(self):
         for handler in logging.getLogger("prompt_manager").handlers[:]:
             handler.close()
+        # Windows refuses to delete an open log: prove every handle is closed.
+        assert_closed(self, self.tmp.name)
         get_logger_manager()._setup_loggers()
         self.tmp.cleanup()
 
@@ -175,15 +193,14 @@ class TestFilesAndRotation(IsolatedLoggerTestCase):
 
     def test_unreadable_files_are_skipped_in_the_listing(self):
         self.logger.info("something")
-        original_stat = pathlib.Path.stat
+        original_lstat = os.lstat
 
-        def stat_failing_for_logs(path, *args, **kwargs):
-            if path.suffix == ".log":
+        def lstat_failing_for_logs(path, *args, **kwargs):
+            if str(path).endswith(".log"):
                 raise OSError("denied")
-            return original_stat(path, *args, **kwargs)
+            return original_lstat(path, *args, **kwargs)
 
-        with mock.patch.object(pathlib.Path, "stat", autospec=True) as stat:
-            stat.side_effect = stat_failing_for_logs
+        with mock.patch.object(logging_config.os, "lstat", lstat_failing_for_logs):
             self.assertEqual(self.manager.get_log_files(), [])
 
     def test_truncate_does_not_roll_the_log_over_while_deleting_backups(self):
@@ -228,10 +245,522 @@ class TestFilesAndRotation(IsolatedLoggerTestCase):
 
     def test_truncate_reports_files_it_cannot_remove(self):
         self._rotate()
-        with mock.patch.object(pathlib.Path, "unlink", side_effect=OSError("busy")):
+
+        def busy(path, *args, **kwargs):
+            raise PermissionError(13, "busy", str(path))
+
+        with mock.patch.object(logging_config.os, "unlink", busy):
             result = self.manager.truncate_logs()
         self.assertEqual(len(result["errors"]), 2)
         self.assertEqual(result["truncated"], ["prompt_manager.log"])
+        for error in result["errors"]:
+            self.assertNotIn(self.tmp.name, error)
+            self.assertRegex(error, r"prompt_manager\.log\.[12]")
+
+
+class TestRotatedCopyRule(IsolatedLoggerTestCase):
+    """Rule R (#329): only prompt_manager.log.<n> regular files are deletable."""
+
+    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)
+        return self._path(name)
+
+    def _symlink(self, name, target_name="secret.txt", data=b"secret " * 1000):
+        """A symlink in the log folder to a file outside it, or None."""
+        target = os.path.join(self.outside, target_name)
+        with open(target, "wb") as handle:
+            handle.write(data)
+        try:
+            os.symlink(target, self._path(name))
+        except (OSError, NotImplementedError):  # pragma: no cover - no privilege
+            return None
+        return target
+
+    def _require_symlink(self, name, **kwargs):
+        target = self._symlink(name, **kwargs)
+        if target is None:  # pragma: no cover - Windows without the privilege
+            self.skipTest("this account cannot create symlinks")
+        return target
+
+    def _kinds(self):
+        return {row["name"]: row["kind"] for row in self.manager.scan_log_folder()}
+
+    def _rotated_rows(self):
+        return [r for r in self.manager.scan_log_folder() if r["kind"] == "rotated"]
+
+    def _names(self):
+        return sorted(os.listdir(self.log_dir))
+
+    def _file_handler(self):
+        return next(
+            h
+            for h in self.manager.logger.handlers
+            if isinstance(h, logging.FileHandler)
+        )
+
+    def test_scan_classes_rotated_active_link_and_other(self):
+        for name in ("prompt_manager.log.1", "prompt_manager.log.2"):
+            self._write(name)
+        self._write("prompt_manager.log.3", b"x" * 3000)
+        for name in (
+            "prompt_manager.log.0",
+            "prompt_manager.log.01",
+            "prompt_manager.log.bak",
+            "prompt_manager.log.1.gz",
+            "prompt_manager.log.-1",
+            "notes.txt",
+        ):
+            self._write(name)
+        os.makedirs(self._path("prompt_manager.log.5"))
+        self._require_symlink("prompt_manager.log.4")
+        kinds = self._kinds()
+        self.assertEqual(
+            kinds,
+            {
+                "prompt_manager.log": "active",
+                "prompt_manager.log.1": "rotated",
+                "prompt_manager.log.2": "rotated",
+                "prompt_manager.log.3": "rotated",
+                "prompt_manager.log.0": "other",
+                "prompt_manager.log.01": "other",
+                "prompt_manager.log.bak": "other",
+                "prompt_manager.log.1.gz": "other",
+                "prompt_manager.log.-1": "other",
+                "notes.txt": "other",
+                "prompt_manager.log.5": "other",
+                "prompt_manager.log.4": "link",
+            },
+        )
+        row = next(
+            r
+            for r in self.manager.scan_log_folder()
+            if r["name"] == "prompt_manager.log.3"
+        )
+        self.assertEqual(
+            set(row), {"name", "kind", "size", "mtime_ns", "ino", "modified"}
+        )
+        stat = os.lstat(self._path("prompt_manager.log.3"))
+        self.assertEqual(row["size"], 3000)
+        self.assertEqual(row["mtime_ns"], stat.st_mtime_ns)
+        self.assertEqual(row["ino"], stat.st_ino)
+
+    def test_scan_matches_names_case_insensitively_on_windows(self):
+        self._write("PROMPT_MANAGER.LOG.2")
+        if os.name == "nt":  # pragma: no cover - native case folding
+            self.assertEqual(self._kinds()["PROMPT_MANAGER.LOG.2"], "rotated")
+            return
+        # POSIX names are case-sensitive: another file, never deleted.
+        self.assertEqual(self._kinds()["PROMPT_MANAGER.LOG.2"], "other")
+        with mock.patch.object(logging_config, "_name_key", ntpath.normcase):
+            self.assertEqual(self._kinds()["PROMPT_MANAGER.LOG.2"], "rotated")
+            result = self.manager.delete_rotated(self._rotated_rows())
+        self.assertEqual(result["removed"], ["PROMPT_MANAGER.LOG.2"])
+
+    def test_a_junction_is_classed_as_a_link_and_kept(self):
+        # Portable control: whatever os.path.isjunction reports is a link.
+        self._write("prompt_manager.log.4")
+        self._write("prompt_manager.log.1")
+        junction = self._path("prompt_manager.log.4")
+        real_isjunction = os.path.isjunction
+
+        def isjunction(path):
+            return os.fspath(path) == junction or real_isjunction(path)
+
+        with mock.patch.object(logging_config.os.path, "isjunction", isjunction):
+            self.assertEqual(self._kinds()["prompt_manager.log.4"], "link")
+            self.manager.delete_rotated(self._rotated_rows())
+            self.manager.truncate_logs()
+        self.assertIn("prompt_manager.log.4", self._names())
+        self.assertNotIn("prompt_manager.log.1", self._names())
+
+    @unittest.skipUnless(os.name == "nt", "junctions exist only on Windows")
+    def test_junction_is_never_deleted_on_windows(self):  # pragma: no cover
+        import _winapi
+
+        target = os.path.join(self.outside, "kept")
+        os.makedirs(target)
+        with open(os.path.join(target, "inside.txt"), "wb") as handle:
+            handle.write(b"keep me")
+        _winapi.CreateJunction(target, self._path("prompt_manager.log.4"))
+        self._write("prompt_manager.log.1")
+        self.assertEqual(self._kinds()["prompt_manager.log.4"], "link")
+        self.manager.delete_rotated(self._rotated_rows())
+        self.manager.truncate_logs()
+        self.assertIn("prompt_manager.log.4", self._names())
+        self.assertTrue(os.path.exists(os.path.join(target, "inside.txt")))
+        with self.assertRaises(FileNotFoundError):
+            self.manager.log_file_entry("prompt_manager.log.4")
+        os.rmdir(self._path("prompt_manager.log.4"))  # removes the junction only
+
+    def test_delete_rotated_removes_exactly_the_listed_files(self):
+        for n in (1, 2, 3):
+            self._write(f"prompt_manager.log.{n}", b"r" * n)
+        self._write("prompt_manager.log.bak")
+        os.makedirs(self._path("prompt_manager.log.7"))
+        target = self._symlink("prompt_manager.log.4")
+        listed = [
+            r for r in self._rotated_rows() if r["name"] != "prompt_manager.log.3"
+        ]
+        result = self.manager.delete_rotated(listed)
+        self.assertEqual(
+            result,
+            {"removed": ["prompt_manager.log.1", "prompt_manager.log.2"], "failed": []},
+        )
+        expected = [
+            "prompt_manager.log",
+            "prompt_manager.log.3",
+            "prompt_manager.log.7",
+            "prompt_manager.log.bak",
+        ]
+        if target is not None:
+            expected.append("prompt_manager.log.4")
+            self.assertTrue(os.path.exists(target))
+        self.assertEqual(self._names(), sorted(expected))
+
+    def test_delete_rotated_refuses_all_when_one_changed(self):
+        def change_size(path):
+            with open(path, "ab") as handle:
+                handle.write(b"more")
+
+        def change_mtime(path):
+            stat = os.lstat(path)
+            os.utime(path, ns=(stat.st_atime_ns, stat.st_mtime_ns + 5_000_000_000))
+
+        def change_inode(path):
+            stat = os.lstat(path)
+            with open(path, "rb") as handle:
+                data = handle.read()
+            other = path + ".new"
+            with open(other, "wb") as handle:
+                handle.write(data)
+            os.utime(other, ns=(stat.st_atime_ns, stat.st_mtime_ns))
+            os.replace(other, path)
+            if os.lstat(path).st_ino in (0, stat.st_ino):  # pragma: no cover
+                self.skipTest("this filesystem reports no inode numbers")
+
+        def remove(path):
+            os.unlink(path)
+
+        def replace_by_link(path):
+            os.unlink(path)
+            target = os.path.join(self.outside, "elsewhere.txt")
+            with open(target, "wb") as handle:
+                handle.write(b"rotated\n")
+            try:
+                os.symlink(target, path)
+            except (OSError, NotImplementedError):  # pragma: no cover
+                self.skipTest("this account cannot create symlinks")
+
+        changes = {
+            "size": change_size,
+            "mtime": change_mtime,
+            "inode": change_inode,
+            "missing": remove,
+            "link": replace_by_link,
+        }
+        for label, change in changes.items():
+            with self.subTest(change=label):
+                for n in (1, 2, 3):
+                    self._write(f"prompt_manager.log.{n}", b"rotated\n")
+                reviewed = self._rotated_rows()
+                change(self._path("prompt_manager.log.2"))
+                before = self._names()
+                with self.assertRaises(RotatedChanged):
+                    self.manager.delete_rotated(reviewed)
+                self.assertEqual(self._names(), before)
+                self.assertIn("prompt_manager.log.1", before)
+                self.assertIn("prompt_manager.log.3", before)
+                for name in self._names():
+                    if name.startswith("prompt_manager.log."):
+                        os.unlink(self._path(name))
+
+    def test_rollover_waits_while_cleanup_holds_the_handler_lock(self):
+        # The file is already over its limit: the next record rolls it over,
+        # renaming .1 to .2 and the active log to .1.
+        self.manager.update_config({"max_file_size": 64, "backup_count": 3})
+        self.logger.info("x" * 100)
+        self._write("prompt_manager.log.1", b"first copy\n")
+        self._write("prompt_manager.log.2", b"second copy\n")
+        reviewed = self._rotated_rows()
+        holding = threading.Event()
+        release = threading.Event()
+        logged = threading.Event()
+        real_unlink = os.unlink
+        self.allowed_thread_names.add("pm-test-rollover")
+
+        def log_from_another_thread():
+            self.logger.info("rolls the log over")
+            logged.set()
+
+        worker = threading.Thread(
+            target=log_from_another_thread, name="pm-test-rollover", daemon=True
+        )
+
+        def slow_unlink(path, *args, **kwargs):
+            if not holding.is_set():
+                worker.start()  # its record must wait for the cleanup's lock
+                holding.set()
+                if not release.wait(10):
+                    raise RuntimeError("the test never released the cleanup")
+            return real_unlink(path, *args, **kwargs)
+
+        result = {}
+
+        def cleanup():
+            result.update(self.manager.delete_rotated(reviewed))
+
+        with mock.patch.object(logging_config.os, "unlink", slow_unlink):
+            cleaner = threading.Thread(target=cleanup, daemon=True)
+            cleaner.start()
+            self.assertTrue(holding.wait(10), "the cleanup never started deleting")
+            # The record waits on the handler lock the cleanup holds.
+            self.assertFalse(logged.wait(0.3), "a record was written mid-cleanup")
+            release.set()
+            cleaner.join(10)
+            worker.join(10)
+        self.assertFalse(cleaner.is_alive())
+        self.assertTrue(logged.is_set())
+        self.assertEqual(
+            result,
+            {"removed": ["prompt_manager.log.1", "prompt_manager.log.2"], "failed": []},
+        )
+        # The rollover ran after the cleanup: only the old active log moved.
+        self.assertEqual(self._names(), ["prompt_manager.log", "prompt_manager.log.1"])
+        with open(self._path("prompt_manager.log.1"), "rb") as handle:
+            self.assertIn(b"x" * 100, handle.read())
+
+    def test_open_rotated_copy_is_reported_not_deleted(self):
+        for n in (1, 2, 3):
+            self._write(f"prompt_manager.log.{n}")
+        real_unlink = os.unlink
+        held = self._path("prompt_manager.log.2")
+
+        def unlink(path, *args, **kwargs):
+            if os.fspath(path) == held:
+                error = PermissionError(
+                    13,
+                    "The process cannot access the file because it is being "
+                    "used by another process",
+                    held,
+                )
+                error.winerror = 32
+                raise error
+            return real_unlink(path, *args, **kwargs)
+
+        with mock.patch.object(logging_config.os, "unlink", unlink):
+            result = self.manager.delete_rotated(self._rotated_rows())
+        self.assertEqual(
+            result["removed"], ["prompt_manager.log.1", "prompt_manager.log.3"]
+        )
+        [(name, error)] = result["failed"]
+        self.assertEqual(name, "prompt_manager.log.2")
+        self.assertIsInstance(error, PermissionError)
+        self.assertIn("prompt_manager.log.2", self._names())
+
+    @unittest.skipUnless(os.name == "nt", "only Windows refuses to unlink an open file")
+    def test_open_rotated_copy_is_reported_not_deleted_on_windows(
+        self,
+    ):  # pragma: no cover
+        for n in (1, 2, 3):
+            self._write(f"prompt_manager.log.{n}")
+        reviewed = self._rotated_rows()
+        handle = open(self._path("prompt_manager.log.2"), "rb")
+        try:
+            result = self.manager.delete_rotated(reviewed)
+        finally:
+            handle.close()
+        self.assertEqual(
+            result["removed"], ["prompt_manager.log.1", "prompt_manager.log.3"]
+        )
+        [(name, error)] = result["failed"]
+        self.assertEqual(name, "prompt_manager.log.2")
+        self.assertIsInstance(error, PermissionError)
+        self.assertIn("prompt_manager.log.2", self._names())
+
+    def test_cleanup_never_logs_while_holding_the_handler_lock(self):
+        for n in (1, 2):
+            self._write(f"prompt_manager.log.{n}")
+        handler = self._file_handler()
+        depth = 0
+        calls = []
+        real_acquire, real_release = handler.acquire, handler.release
+
+        def acquire():
+            nonlocal depth
+            real_acquire()
+            depth += 1
+
+        def release():
+            nonlocal depth
+            depth -= 1
+            real_release()
+
+        real_call_handlers = logging.Logger.callHandlers
+
+        def call_handlers(logger_self, record):
+            calls.append((record.getMessage(), depth))
+            return real_call_handlers(logger_self, record)
+
+        with (
+            mock.patch.object(handler, "acquire", acquire),
+            mock.patch.object(handler, "release", release),
+            mock.patch.object(logging.Logger, "callHandlers", call_handlers),
+        ):
+            self.manager.delete_rotated(self._rotated_rows())
+            self.assertEqual(depth, 0)
+            self._write("prompt_manager.log.1")
+            self.manager.truncate_logs()
+            self.assertEqual(depth, 0)
+        self.assertTrue(calls, "the cleanup logged no summary")
+        self.assertEqual([c for c in calls if c[1]], [])
+
+    def test_truncate_keeps_other_files_and_links(self):
+        # Corrected (#329): the legacy glob deleted every prompt_manager.log*.
+        self._rotate()
+        self._write("prompt_manager.log.bak", b"mine")
+        self._write("prompt_manager.log.1.gz", b"zip")
+        target = self._symlink("prompt_manager.log.4")
+        result = self.manager.truncate_logs()
+        self.assertEqual(result["errors"], [])
+        self.assertEqual(
+            sorted(result["truncated"]),
+            ["prompt_manager.log", "prompt_manager.log.1", "prompt_manager.log.2"],
+        )
+        names = self._names()
+        self.assertIn("prompt_manager.log.bak", names)
+        self.assertIn("prompt_manager.log.1.gz", names)
+        self.assertNotIn("prompt_manager.log.1", names)
+        if target is not None:
+            self.assertIn("prompt_manager.log.4", names)
+            self.assertTrue(os.path.islink(self._path("prompt_manager.log.4")))
+            with open(target, "rb") as handle:
+                self.assertEqual(handle.read(), b"secret " * 1000)
+
+    @unittest.skipIf(os.name == "nt", "Windows cannot rename a log that is open")
+    def test_truncate_never_empties_a_renamed_log_the_handler_still_writes(self):
+        # The user renamed the active log to keep it; the handler still has it
+        # open. Clear must empty the new prompt_manager.log, not the kept copy.
+        self.logger.info("worth keeping")
+        kept = os.path.join(self.log_dir, "saved_copy.txt")
+        os.rename(self._path("prompt_manager.log"), kept)
+        with open(kept, "rb") as handle:
+            before = handle.read()
+        self._write("prompt_manager.log", b"fresh active log\n")
+        result = self.manager.truncate_logs()
+        self.assertEqual(result["errors"], [])
+        with open(kept, "rb") as handle:
+            self.assertTrue(handle.read().startswith(before))
+        with open(self._path("prompt_manager.log"), "rb") as handle:
+            self.assertEqual(handle.read(), b"")
+
+    def test_name_surrogate_reparse_points_are_links_and_placeholders_are_not(self):
+        regular = stat.S_IFREG | 0o644
+        path = self._write("prompt_manager.log.3")
+        for tag, is_link in (
+            (0xA000001D, True),  # WSL symlink
+            (0xA0000003, True),  # mount point (junction)
+            (0x9000001A, False),  # OneDrive cloud placeholder: a regular file
+            (0, False),
+        ):
+            with self.subTest(tag=hex(tag)):
+                st = mock.Mock(st_mode=regular, st_reparse_tag=tag)
+                self.assertIs(logging_config._is_link(path, st), is_link)
+
+    def test_truncate_with_file_logging_off_never_creates_the_active_log(self):
+        self.manager.update_config({"file_logging": False})
+        os.unlink(self._path("prompt_manager.log"))
+        self._write("prompt_manager.log.1")
+        result = self.manager.truncate_logs()
+        self.assertEqual(result, {"truncated": ["prompt_manager.log.1"], "errors": []})
+        self.assertEqual(self._names(), [])
+
+    def test_truncate_never_empties_a_linked_active_log(self):
+        self.manager.update_config({"file_logging": False})
+        os.unlink(self._path("prompt_manager.log"))
+        target = self._require_symlink("prompt_manager.log", data=b"keep")
+        self.manager.truncate_logs()
+        with open(target, "rb") as handle:
+            self.assertEqual(handle.read(), b"keep")
+
+    def test_log_files_mark_kind_and_lstat_links(self):
+        self._write("prompt_manager.log.1", b"r" * 10)
+        self._write("prompt_manager.log.bak", b"b")
+        target = self._require_symlink("prompt_manager.log.4")
+        rows = {row["filename"]: row for row in self.manager.get_log_files()}
+        self.assertEqual(rows["prompt_manager.log"]["kind"], "active")
+        self.assertTrue(rows["prompt_manager.log"]["is_main"])
+        self.assertEqual(rows["prompt_manager.log.1"]["kind"], "rotated")
+        self.assertEqual(rows["prompt_manager.log.bak"]["kind"], "other")
+        link = rows["prompt_manager.log.4"]
+        self.assertEqual(link["kind"], "link")
+        self.assertEqual(
+            link["size"], os.lstat(self._path("prompt_manager.log.4")).st_size
+        )
+        self.assertNotEqual(link["size"], os.path.getsize(target))
+        for row in rows.values():
+            self.assertEqual(
+                set(row), {"filename", "path", "size", "modified", "is_main", "kind"}
+            )
+
+    def test_an_unreadable_timestamp_does_not_break_the_scan(self):
+        self._write("prompt_manager.log.1")
+        real = logging_config.datetime
+
+        class Clock:
+            @staticmethod
+            def fromtimestamp(value):
+                raise OSError(22, "Invalid argument")
+
+        with mock.patch.object(logging_config, "datetime", Clock):
+            rows = {r["name"]: r for r in self.manager.scan_log_folder()}
+        self.assertIs(logging_config.datetime, real)
+        self.assertEqual(rows["prompt_manager.log.1"]["kind"], "rotated")
+        self.assertEqual(rows["prompt_manager.log.1"]["modified"], "")
+
+    def test_stats_exclude_links(self):
+        self._write("prompt_manager.log.1", b"r" * 10)
+        before = self.manager.get_log_stats()
+        self._require_symlink("prompt_manager.log.4")
+        after = self.manager.get_log_stats()
+        self.assertEqual(after["log_files_count"], before["log_files_count"])
+        self.assertEqual(after["total_log_size"], before["total_log_size"])
+
+    def test_log_file_entry_accepts_only_listed_regular_files(self):
+        self._write("prompt_manager.log.1")
+        self._write("prompt_manager.log.bak")
+        os.makedirs(self._path("folder.log"))
+        self._symlink("prompt_manager.log.4")
+        self.assertEqual(
+            os.path.normcase(str(self.manager.log_file_entry("prompt_manager.log.1"))),
+            os.path.normcase(self._path("prompt_manager.log.1")),
+        )
+        self.manager.log_file_entry("prompt_manager.log")
+        self.manager.log_file_entry("prompt_manager.log.bak")
+        for name in (
+            "prompt_manager.log.4",
+            "folder.log",
+            "missing.log",
+            "",
+            ".",
+            "..",
+            "../outside/secret.txt",
+            "..\\outside\\secret.txt",
+            "prompt_manager.log:stream",
+            "prompt_manager.log\x00",
+        ):
+            with self.subTest(name=name):
+                with self.assertRaises(FileNotFoundError):
+                    self.manager.log_file_entry(name)
 
 
 class TestConfiguration(IsolatedLoggerTestCase):

tests/vendor/browser-image.test.mjs

@@ -28,24 +28,3 @@ for (const name of ['frontend-coverage', 'browser-workflows']) {
         assert.match(body, /node --version \| grep -q '\^v24\\\.'/, 'the image Node is the Node 24 the project requires');
     });
 }
-
-// The real-host job drives Chromium against a managed ComfyUI. `playwright install --with-deps` pulled the browser's
-// system packages from Ubuntu's apt mirror on every run, and three runs (#349, #378, #382) crawled at ~26 kB/s until
-// the job hit its limit after every test had passed. The pinned image ships Chromium and its libraries, so the job runs
-// there; the image Python must be the reviewed host Python from tools/vendor/comfyui-runtime.json.
-test('real ComfyUI compatibility runs in the pinned Playwright image on the reviewed host Python', () => {
-    const testWorkflow = readFileSync(new URL('../../.github/workflows/test.yml', import.meta.url), 'utf8');
-    const runtime = JSON.parse(readFileSync(new URL('../../tools/vendor/comfyui-runtime.json', import.meta.url), 'utf8'));
-    const match = testWorkflow.match(/^ {2}comfyui:\n([\s\S]*?)(?=^ {2}[a-z][a-z-]*:\n|(?![\s\S]))/m);
-    assert.ok(match, 'comfyui job exists');
-    const body = match[1];
-    assert.match(body, new RegExp(`^    container:\\n      image: mcr\\.microsoft\\.com/playwright:v${version.replaceAll('.', '\\.')}-noble$`, 'm'));
-    assert.doesNotMatch(body, /playwright install/, 'Chromium and its system libraries come from the image, not apt');
-    assert.doesNotMatch(body, /apt-get|actions\/setup-(python|node)@/);
-    assert.match(body, /python3 -m venv --without-pip /, 'a user-space venv on the image Python');
-    assert.match(body, /sha256sum -c/, 'the pip wheel that bootstraps the venv is checksum-pinned');
-    assert.match(body, /node --version \| grep -q '\^v24\\\.'/, 'the image Node is Node 24');
-    // The image Python is checked against the pin at run time, so a pin or image bump that disagrees fails loudly.
-    assert.match(body, /steps\.runtime\.outputs\.pythonVersion/, 'the job compares the image Python with the reviewed pin');
-    assert.match(runtime.pythonVersion, /^3\.12$/, 'the reviewed host Python is the image Python (3.12 on noble)');
-});

utils/image_monitor.py

@@ -1075,9 +1075,7 @@ class ImageMonitor:
         """Stop the image monitoring system within one ``timeout`` deadline.
 
         Closes the handler first (its worker is joined and its dispatch
-        returns at once), then stops the watchdog observer and joins it. The
-        worker and the observer share the one deadline: the observer's join
-        gets what the worker left, possibly 0 s.
+        returns at once), then stops the watchdog observer and joins it.
         watchdog's ``observer.stop()`` joins its emitters without a limit;
         it counts against the deadline but is not bounded here. A part an
         earlier call left running is waited for again. Safe to call when
@@ -1102,6 +1100,10 @@ class ImageMonitor:
         for pair_observer, pair_handler in pairs:
             handler_ok = self._close_handler(pair_handler, remaining())
             observer_ok = self._stop_observer(pair_observer, remaining())
+            if not observer_ok:
+                self.logger.warning(
+                    f"watchdog observer still running after {timeout} s"
+                )
             if not (handler_ok and observer_ok):
                 self._unfinished.append(
                     (
@@ -1122,8 +1124,7 @@ class ImageMonitor:
             return False
 
     def _stop_observer(self, observer, timeout: float) -> bool:
-        """``observer.stop()`` (safe to repeat), then a join of ``timeout``
-        seconds; a join that times out is logged with the time it got."""
+        """``observer.stop()`` (safe to repeat), then a bounded join."""
         if observer is None:
             return True
         try:
@@ -1132,12 +1133,7 @@ class ImageMonitor:
         except Exception as e:
             self.logger.warning(f"Error while stopping image monitoring: {e}")
             return False
-        if observer.is_alive():
-            self.logger.warning(
-                f"watchdog observer still running after {timeout:.1f} s"
-            )
-            return False
-        return True
+        return not observer.is_alive()
 
     def detect_comfyui_output_dirs(self) -> list:
         """Auto-detect ComfyUI output directories.

utils/logging_config.py

@@ -10,9 +10,12 @@ This module provides centralized logging configuration with support for:
 """
 
 import collections
+import fnmatch
 import logging
 import logging.handlers
 import os
+import re
+import stat
 import threading
 from datetime import datetime
 from typing import Dict, Any, Optional, List
@@ -20,6 +23,160 @@ from pathlib import Path
 
 from . import library_location
 
+ACTIVE_LOG_NAME = "prompt_manager.log"
+# Rule R (#329): a rotated copy is prompt_manager.log.<n>, n >= 1, no leading 0.
+_ROTATED_NAME = re.compile(r"prompt_manager\.log\.([1-9][0-9]*)")
+# The legacy listing's pattern (GET /logs/files and /logs/stats).
+_LISTED_PATTERN = "*.log*"
+LOG_FILE_KINDS = ("active", "rotated", "link", "other")
+
+
+class RotatedChanged(Exception):
+    """A reviewed rotated copy changed since the review; nothing was deleted."""
+
+    def __init__(self, name):
+        super().__init__(f"{name} changed since the review")
+        self.name = name
+
+
+def _name_key(name: str) -> str:
+    """How a name compares: case-folded where the filesystem folds it (Windows)."""
+    return os.path.normcase(name)
+
+
+def _rotated_number(name: str) -> Optional[int]:
+    match = _ROTATED_NAME.fullmatch(_name_key(name))
+    return int(match.group(1)) if match else None
+
+
+# Windows reparse tags with the name-surrogate bit name another file or
+# folder (symlinks, junctions, WSL symlinks); cloud placeholders do not.
+_NAME_SURROGATE = 0x20000000
+
+
+def _is_link(path: str, st: os.stat_result) -> bool:
+    if getattr(st, "st_reparse_tag", 0) & _NAME_SURROGATE:
+        return True
+    return stat.S_ISLNK(st.st_mode) or os.path.isjunction(path)
+
+
+class LogFilePath(type(Path())):
+    """A listed log file's path, with the inode it had when it was listed."""
+
+    listed_ino = 0
+
+
+def _kind(name: str, path: str, st: os.stat_result) -> str:
+    """Rule R: ``active``, ``rotated``, ``link`` or ``other`` (never deleted)."""
+    if _is_link(path, st):
+        return "link"
+    if not stat.S_ISREG(st.st_mode):
+        return "other"
+    if _name_key(name) == _name_key(ACTIVE_LOG_NAME):
+        return "active"
+    if _rotated_number(name) is not None:
+        return "rotated"
+    return "other"
+
+
+def scan_log_folder(log_dir) -> List[Dict[str, Any]]:
+    """Every direct entry of ``log_dir`` with its Rule R kind (blocking).
+
+    Rows are ``{name, kind, size, mtime_ns, ino, modified}``, read with
+    ``lstat`` so a link is described, never followed. An entry that cannot
+    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
+
+
+def public_row(row: Dict[str, Any]) -> Dict[str, Any]:
+    """A scan row without its private keys."""
+    return {key: value for key, value in row.items() if not key.startswith("_")}
+
+
+def rotated_sort_key(row: Dict[str, Any]) -> int:
+    return _rotated_number(row["name"]) or 0
+
+
+def log_file_entry(log_dir, name) -> Path:
+    """The path of a downloadable log file named ``name`` (blocking).
+
+    Only a regular, non-link file listed directly in ``log_dir`` under that
+    exact name; a separator, ``..``, a stream suffix or any unlisted name
+    raises FileNotFoundError, as a missing file does.
+    """
+    if not isinstance(name, str) or not name or name in (".", ".."):
+        raise FileNotFoundError("Log file not found")
+    if any(ch in name for ch in ("/", "\\", "\x00", ":")):
+        raise FileNotFoundError("Log file not found")
+    for row in scan_log_folder(log_dir):
+        if row["name"] == name and row["kind"] in ("active", "rotated", "other"):
+            if not row["_regular"]:  # a folder, FIFO, socket or device
+                break
+            entry = LogFilePath(log_dir, row["name"])
+            entry.listed_ino = row["ino"]
+            return entry
+    raise FileNotFoundError("Log file not found")
+
+
+def _stream_is(stream, ino: int) -> bool:
+    """Whether ``stream`` is open on the file with inode ``ino`` (0: unknown)."""
+    try:
+        open_ino = os.fstat(stream.fileno()).st_ino
+    except (OSError, ValueError, AttributeError):
+        return False
+    return not (ino and open_ino and open_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):
+        parts = [error.strerror or type(error).__name__]
+        if error.filename:
+            parts.append(os.path.basename(str(error.filename)))
+        return ": ".join(parts)
+    return str(error) or type(error).__name__
+
+
+def _matches_review(row: Optional[Dict[str, Any]], reviewed: Dict[str, Any]) -> bool:
+    if row is None or row["kind"] != "rotated":
+        return False
+    if row["size"] != reviewed.get("size") or row["mtime_ns"] != reviewed.get(
+        "mtime_ns"
+    ):
+        return False
+    ino = reviewed.get("ino") or 0
+    return not (ino and row["ino"] and row["ino"] != ino)
+
 
 class _OwnedRotatingFileHandler(logging.handlers.RotatingFileHandler):
     """Keep a retired owned sink closed when an in-flight record resumes."""
@@ -87,6 +244,9 @@ class PromptManagerLogger:
         # Memory buffer for recent logs (for web viewer)
         self._log_buffer = collections.deque(maxlen=self.config["buffer_size"])
         self._buffer_lock = threading.Lock()
+        # Held while a handler is replaced or log files are deleted (#329):
+        # taken before any file handler's own lock, never the other way round.
+        self._files_lock = threading.Lock()
 
         # Initialize loggers
         self._setup_loggers()
@@ -238,6 +398,14 @@ class PromptManagerLogger:
         # Return most recent first
         return list(reversed(logs[-limit:]))
 
+    def scan_log_folder(self) -> List[Dict[str, Any]]:
+        """Every entry of the log folder with its Rule R kind (blocking)."""
+        return [public_row(row) for row in scan_log_folder(self.log_dir)]
+
+    def log_file_entry(self, name: str) -> Path:
+        """A downloadable (regular, non-link) log file; FileNotFoundError else."""
+        return log_file_entry(self.log_dir, name)
+
     def get_log_files(self) -> List[Dict[str, Any]]:
         """Get information about available log files.
 
@@ -245,26 +413,27 @@ class PromptManagerLogger:
             List of dictionaries containing file information:
             - filename: Name of the log file
             - path: Full path to the log file
-            - size: File size in bytes
+            - size: File size in bytes (a link's own size: links are not followed)
             - modified: ISO formatted modification timestamp
             - is_main: True if this is the main log file
+            - kind: ``active``, ``rotated``, ``link`` or ``other`` (Rule R)
         """
         log_files = []
-
-        for log_file in self.log_dir.glob("*.log*"):
-            try:
-                stat = log_file.stat()
-                log_files.append(
-                    {
-                        "filename": log_file.name,
-                        "path": str(log_file),
-                        "size": stat.st_size,
-                        "modified": datetime.fromtimestamp(stat.st_mtime).isoformat(),
-                        "is_main": log_file.name == "prompt_manager.log",
-                    }
-                )
-            except Exception as e:
-                self.logger.error(f"Error getting info for log file {log_file}: {e}")
+        for row in scan_log_folder(self.log_dir):
+            if row["_dir"]:
+                continue
+            if not fnmatch.fnmatchcase(_name_key(row["name"]), _LISTED_PATTERN):
+                continue
+            log_files.append(
+                {
+                    "filename": row["name"],
+                    "path": str(self.log_dir / row["name"]),
+                    "size": row["size"],
+                    "modified": row["modified"],
+                    "is_main": row["kind"] == "active",
+                    "kind": row["kind"],
+                }
+            )
 
         # Sort by modification time, newest first
         log_files.sort(key=lambda x: x["modified"], reverse=True)
@@ -302,14 +471,44 @@ class PromptManagerLogger:
             self.logger.error(f"Error reading log file {filename}: {e}")
             raise
 
-    def _truncate_main_log(self, log_file: Path) -> None:
+    def _file_handlers(self) -> List[logging.Handler]:
+        return [h for h in self.logger.handlers if isinstance(h, logging.FileHandler)]
+
+    def _acquire_file_handlers(self) -> List[logging.Handler]:
+        """Take every file handler's lock: no rollover renames a file meanwhile.
+
+        The caller holds ``_files_lock`` (so no handler is replaced) and must
+        not log until it released them: the lock is re-entrant, so a record
+        logged here could roll the log over under the caller's feet.
+        """
+        held = []
+        try:
+            for handler in self._file_handlers():
+                handler.acquire()
+                held.append(handler)
+        except BaseException:
+            self._release_handlers(held)
+            raise
+        return held
+
+    @staticmethod
+    def _release_handlers(held) -> None:
+        for handler in reversed(held):
+            handler.release()
+
+    def _truncate_main_log(self, log_file: Path, expected_ino: int = 0) -> None:
         """Empty the main log through the handler that writes to it.
 
         Opening the file a second time with ``"w"`` leaves the handler's own
         stream at its old offset: later records land past the new end
         (zero-padded on POSIX) and ``tell()`` keeps reporting the old size, so
         the next record rolls the file over. Truncating the handler's stream
-        keeps position and size in step.
+        keeps position and size in step. Without a handler on it (file logging
+        off) the existing file is emptied in place, never through a link.
+
+        A handler whose open stream is not the listed file (``expected_ino``:
+        the user renamed the log, which the handler still writes to) is left
+        alone: the file under the active name is emptied instead.
         """
         handled = False
         target = os.path.normcase(os.path.realpath(str(log_file)))
@@ -320,6 +519,8 @@ class PromptManagerLogger:
                 continue
             if os.path.normcase(os.path.realpath(base)) != target:
                 continue
+            if not _stream_is(stream, expected_ino):
+                continue
             handler.acquire()
             try:
                 stream.flush()
@@ -329,14 +530,65 @@ class PromptManagerLogger:
             finally:
                 handler.release()
         if not handled:
-            with open(log_file, "w", encoding="utf-8", errors="replace") as f:
-                f.write("")
+            flags = os.O_WRONLY | os.O_TRUNC | getattr(os, "O_NOFOLLOW", 0)
+            os.close(os.open(str(log_file), flags))
+
+    def delete_rotated(self, expected: List[Dict[str, Any]]) -> Dict[str, Any]:
+        """Delete exactly the reviewed rotated copies, or none of them (blocking).
+
+        ``expected`` are scan rows (``name``, ``size``, ``mtime_ns``,
+        ``ino``). Under ``_files_lock`` and every file handler's lock the
+        folder is scanned again; when any expected file is missing, is no
+        longer a rotated copy (Rule R) or differs in size, ``mtime_ns`` or a
+        non-zero ``ino``, RotatedChanged is raised with nothing deleted.
+        Otherwise each is unlinked in ``<n>`` order.
+
+        Returns ``{"removed": [names], "failed": [(name, OSError)]}``. One
+        summary line is logged once both locks are released.
+        """
+        removed, failed = [], []
+        changed = None
+        with self._files_lock:
+            held = self._acquire_file_handlers()
+            try:
+                current = {
+                    _name_key(row["name"]): row for row in scan_log_folder(self.log_dir)
+                }
+                targets = []
+                for reviewed in sorted(expected, key=rotated_sort_key):
+                    row = current.get(_name_key(str(reviewed.get("name", ""))))
+                    if not _matches_review(row, reviewed):
+                        changed = str(reviewed.get("name", ""))
+                        break
+                    targets.append(row["name"])
+                if changed is None:
+                    for name in targets:
+                        try:
+                            os.unlink(os.path.join(self.log_dir, name))
+                            removed.append(name)
+                        except OSError as error:
+                            failed.append((name, error))
+            finally:
+                self._release_handlers(held)
+        if changed is not None:
+            self.logger.warning(
+                "Rotated log cleanup refused: %s changed since the review", changed
+            )
+            raise RotatedChanged(changed)
+        self.logger.info(
+            "Deleted %d rotated log copies%s",
+            len(removed),
+            f", {len(failed)} could not be deleted" if failed else "",
+        )
+        return {"removed": removed, "failed": failed}
 
     def truncate_logs(self) -> Dict[str, Any]:
-        """Truncate all log files.
+        """Empty the active log and delete its rotated copies (Rule R).
 
-        Clears the main log file and deletes rotated log files.
-        Also clears the memory buffer.
+        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.
 
         Returns:
             Dictionary with:
@@ -345,28 +597,31 @@ class PromptManagerLogger:
         """
         results = {"truncated": [], "errors": []}
 
-        # Collect first and stay silent until every file has been handled:
-        # a log line written mid-way can roll the just-emptied main log over
-        # and recreate the backup that was deleted a moment earlier.
-        log_files = sorted(self.log_dir.glob("prompt_manager.log*"))
-        for log_file in log_files:
+        with self._files_lock:
+            held = self._acquire_file_handlers()
             try:
-                if log_file.name == "prompt_manager.log":
-                    self._truncate_main_log(log_file)
-                else:
-                    # For rotated files, delete them
-                    log_file.unlink()
-
-                results["truncated"].append(log_file.name)
-
-            except Exception as e:
-                results["errors"].append(
-                    f"Failed to truncate {log_file.name}: {str(e)}"
+                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,
                 )
-
-        # Clear memory buffer
-        with self._buffer_lock:
-            self._log_buffer.clear()
+                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()
+            finally:
+                self._release_handlers(held)
 
         if results["truncated"]:
             self.logger.info("Truncated log files: %s", ", ".join(results["truncated"]))
@@ -409,7 +664,9 @@ class PromptManagerLogger:
                 "backup_count",
             ]
         ):
-            self._setup_loggers()
+            # No cleanup may hold the old handler while it is replaced.
+            with self._files_lock:
+                self._setup_loggers()
 
         self.logger.info(f"Updated logging configuration: {new_config}")
 
@@ -440,7 +697,9 @@ class PromptManagerLogger:
                 level = entry["level"]
                 level_counts[level] = level_counts.get(level, 0) + 1
 
-        log_files = self.get_log_files()
+        # A link's target is not a log file of this folder: neither counted
+        # nor sized.
+        log_files = [f for f in self.get_log_files() if f["kind"] != "link"]
         total_log_size = sum(f["size"] for f in log_files)
 
         return {

utils/version.py

@@ -0,0 +1,18 @@
+"""The project version, read from ``pyproject.toml`` (#329)."""
+
+import re
+from pathlib import Path
+
+
+def project_version(root):
+    """The version in ``root``'s ``pyproject.toml``; ``"unknown"`` on failure."""
+    try:
+        pyproject_path = Path(root) / "pyproject.toml"
+        if pyproject_path.exists():
+            content = pyproject_path.read_text()
+            match = re.search(r'version\s*=\s*["\']([^"\']+)["\']', content)
+            if match:
+                return match.group(1)
+    except Exception:  # nosec B110 - an unreadable file reads as unknown
+        pass
+    return "unknown"