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

# Conflicts:
#	tests/browser/workspace_system_fixture.py

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,20 +23,24 @@ Features:
 """
 
 import atexit
+import re
 import threading
 import time
 from pathlib import Path
 
 
 def get_version():
-    """Parse version from pyproject.toml (``utils/version.py``)."""
-    from .utils.version import project_version
-
+    """Parse version from pyproject.toml"""
     try:
-        root = Path(__file__).parent
+        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)
     except Exception:
-        return "unknown"
-    return project_version(root)
+        pass
+    return "unknown"
 
 
 from .prompt_manager import (  # noqa: E402
@@ -264,89 +268,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,8 +5,6 @@ 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
@@ -67,8 +65,6 @@ 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
@@ -327,8 +323,6 @@ class PromptManagerAPI(
     WorkflowStorageRoutesMixin,
     DuplicateFilesMixin,
     LoggingRoutesMixin,
-    LogCleanupMixin,
-    DiagnosticsReportMixin,
     AutotagRoutesMixin,
     LoraIntegrationMixin,
     LoraImportRoutesMixin,
@@ -684,7 +678,6 @@ 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,29 +489,9 @@ def _diagnose_dependencies():
     }
 
 
-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):
+def _diagnose_output_dirs(configured_dirs):
     """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
@@ -545,10 +525,9 @@ def _diagnose_output_dirs(configured_dirs, public_path=None):
     }
 
 
-def _diagnose_image_monitor(public_path=None):
+def _diagnose_image_monitor():
     """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:
@@ -1076,7 +1055,16 @@ class AdminRoutesMixin:
     async def run_diagnostics(self, request):
         """Run comprehensive system diagnostics and health checks."""
         try:
-            results = legacy_checks(self.db.model.db_path)
+            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(),
+            }
             return web.json_response({"success": True, "diagnostics": results})
 
         except Exception as e:

py/api/diagnostics_report.py

@@ -1,289 +0,0 @@
-"""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

@@ -1,207 +0,0 @@
-"""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,13 +1,6 @@
-"""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.
-"""
+"""Logging API routes for PromptManager."""
 
 import os
-import stat
 
 from aiohttp import web
 
@@ -92,41 +85,6 @@ 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."""
 
@@ -237,16 +195,16 @@ class LoggingRoutesMixin:
                     {"success": False, "error": "Invalid filename"}, status=400
                 )
 
-            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.
+            log_file_path = logger_manager.log_dir / filename
+
+            if not log_file_path.exists():
                 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",
@@ -266,7 +224,7 @@ class LoggingRoutesMixin:
         """Truncate all log files."""
         try:
             logger_manager = self._get_logger_manager()
-            results = await self._run_in_executor(logger_manager.truncate_logs)
+            results = logger_manager.truncate_logs()
 
             return web.json_response(
                 {
@@ -308,7 +266,7 @@ class LoggingRoutesMixin:
                 return web.json_response({"success": False, "error": error}, status=400)
 
             logger_manager = self._get_logger_manager()
-            await self._run_in_executor(logger_manager.update_config, clean)
+            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,9 +2,7 @@
 
 ``POST /prompt_manager/maintenance/preview {operation, options:{}}`` reviews
 one operation (``duplicate_prompts``, ``optimize``, ``unused_components`` or
-``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
+``missing_links``; ``database/maintenance.py``) 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
@@ -56,7 +54,6 @@ 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:
@@ -86,8 +83,7 @@ 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
-# rotated_logs (#329, py/api/log_cleanup.py) reviews log files, not records.
-PREVIEW_OPERATIONS = MAINTENANCE_OPERATIONS + (DUPLICATE_FILES, ROTATED_LOGS)
+PREVIEW_OPERATIONS = MAINTENANCE_OPERATIONS + (DUPLICATE_FILES,)
 # 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.
@@ -395,8 +391,6 @@ 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:
@@ -498,17 +492,6 @@ 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/fast_reset.py

@@ -1,283 +0,0 @@
-"""Run a browser-test fixture seeder off the live database, then publish once (#384).
-
-The seeders make hundreds of small commits. On a slow CI disk the live
-database's commits queue behind WAL checkpoint syncs and a reset took over
-30 s. ``ScratchSeeder.run`` instead:
-
-1. copies the live database into a scratch database (SQLite backup API) whose
-   connections never sync (a crash loses only the copy);
-2. runs the unchanged seeder against an API proxy whose ``db`` is the scratch
-   database; everything else (files, thumbnails, caches, command state) is
-   the live API's, written exactly as before;
-3. replaces the rows of every table the seed changed in the live database, in
-   one transaction on the live connection. Tables the seed left alone are not
-   touched, so state another fixture owns stays as it was.
-
-A commit by any other live connection while the seed ran fails the reset
-instead of being overwritten (``PRAGMA data_version``).
-"""
-
-from __future__ import annotations
-
-import hashlib
-import os
-import shutil
-import sqlite3
-import tempfile
-import weakref
-from pathlib import Path
-from typing import Any, Callable, TypeVar
-
-T = TypeVar("T")
-
-# Set to 1 to run seeders on the live database as before #384 (the equivalence
-# proof, tests/browser/reset_equivalence.py, compares both ways).
-SEED_ON_LIVE_ENV = "PM_TEST_SEED_ON_LIVE"
-SCRATCH_SCHEMA = "reset_seed"
-# API calls a seed makes that must see the published rows (see _SeedApi).
-DEFERRED_CALLS = frozenset({"invalidate_gallery_cache"})
-# Linux's shared-memory tmpfs (present in the Playwright image and on runners).
-MEMORY_ROOT = Path("/dev/shm")
-# The largest fixtures' scratch copy is about 8 MB (database plus a WAL that
-# autocheckpoints near 4 MB). Docker's default /dev/shm is 64 MB, shared with
-# the browsers and the other workers; with less free than this, the scratch
-# copy goes on disk instead.
-MEMORY_FREE_BYTES = 32 * 1024 * 1024
-# Scratch folders are named <prefix><owner pid>-<random>.
-SCRATCH_PREFIX = "pm-reset-"
-
-
-def _quote(name: str) -> str:
-    return '"' + name.replace('"', '""') + '"'
-
-
-def _tables(conn: sqlite3.Connection, schema: str = "main") -> list[str]:
-    rows = conn.execute(
-        f"SELECT name, sql FROM {schema}.sqlite_master WHERE type = 'table' "
-        "ORDER BY name"
-    ).fetchall()
-    for name, sql in rows:
-        if sql and sql.lstrip().upper().startswith("CREATE VIRTUAL"):
-            raise NotImplementedError(f"virtual table {name} cannot be copied")
-    return [name for name, _sql in rows]
-
-
-def _columns(conn: sqlite3.Connection, table: str, schema: str = "main") -> list[str]:
-    """Writable columns (hidden and generated ones excluded), plus ``rowid``
-    when the table has a rowid that no INTEGER PRIMARY KEY column carries."""
-    info = conn.execute(f"PRAGMA {schema}.table_xinfo({_quote(table)})").fetchall()
-    columns = [row[1] for row in info if row[6] == 0]
-    keys = [row for row in info if row[5]]
-    sql = conn.execute(
-        f"SELECT sql FROM {schema}.sqlite_master WHERE type = 'table' AND name = ?",
-        (table,),
-    ).fetchone()[0]
-    without_rowid = sql is not None and "WITHOUT ROWID" in sql.upper()
-    integer_key = len(keys) == 1 and str(keys[0][2]).upper() == "INTEGER"
-    if not without_rowid and not integer_key and table != "sqlite_sequence":
-        columns.insert(0, "rowid")
-    return columns
-
-
-def table_digests(conn: sqlite3.Connection, schema: str = "main") -> dict[str, str]:
-    """A digest of every table's rows (rowids included), by table name."""
-    result = {}
-    for table in _tables(conn, schema):
-        selected = ", ".join(
-            _quote(c) if c != "rowid" else "rowid"
-            for c in _columns(conn, table, schema)
-        )
-        digest = hashlib.sha256()
-        rows = conn.execute(f"SELECT {selected} FROM {schema}.{_quote(table)}")
-        for row in sorted(repr(tuple(row)) for row in rows):
-            digest.update(row.encode("utf-8", "surrogatepass"))
-            digest.update(b"\n")
-        result[table] = digest.hexdigest()
-    return result
-
-
-def _process_alive(pid: int) -> bool:
-    try:
-        os.kill(pid, 0)
-    except ProcessLookupError:
-        return False
-    except OSError:
-        return True  # it exists but belongs to someone else
-    return True
-
-
-def remove_orphaned_scratch(root: Path) -> None:
-    """Remove scratch folders of servers that died without cleaning up.
-
-    A server killed outright (SIGKILL, a crash) never runs ``close``, and
-    tmpfs keeps its folder in memory until reboot. The folder name carries
-    the owner's process id; a folder whose process is gone is removed.
-    """
-    if os.name != "posix":
-        return  # os.kill(pid, 0) would terminate the process on Windows
-    for folder in root.glob(f"{SCRATCH_PREFIX}*-*"):
-        pid = folder.name[len(SCRATCH_PREFIX) :].split("-", 1)[0]
-        if pid.isdigit() and not _process_alive(int(pid)):
-            shutil.rmtree(folder, ignore_errors=True)
-
-
-class _SeedApi:
-    """The live API, except that ``db`` is the scratch database.
-
-    Cache drops the seed asks for (``DEFERRED_CALLS``) are recorded and made
-    after the publish: dropped earlier, a request in between would cache the
-    live database's old rows again.
-    """
-
-    def __init__(self, api: Any, db: Any) -> None:
-        object.__setattr__(self, "_api", api)
-        object.__setattr__(self, "db", db)
-        object.__setattr__(self, "deferred", [])
-
-    def __getattr__(self, name: str) -> Any:
-        if name in DEFERRED_CALLS:
-
-            def record(*args: Any, **kwargs: Any) -> None:
-                self.deferred.append((name, args, kwargs))
-
-            return record
-        return getattr(self._api, name)
-
-    def __setattr__(self, name: str, value: Any) -> None:
-        setattr(self._api, name, value)
-
-
-class ScratchSeeder:
-    """Seed into a scratch copy and publish the changed tables in one commit."""
-
-    def __init__(
-        self,
-        api: Any,
-        directory: Path,
-        *,
-        on_live: bool = False,
-        memory_root: Path | None = MEMORY_ROOT,
-    ) -> None:
-        self._api = api
-        self._fallback = Path(directory)
-        self._memory_root = memory_root
-        self._directory: Path | None = None
-        self._on_live = on_live
-        self._scratch = None
-        self._unsynced: "weakref.WeakSet[sqlite3.Connection]" = weakref.WeakSet()
-
-    def run(self, seed: Callable[[Any], T]) -> T:
-        """Run ``seed(api)`` (blocking; call it in an executor thread)."""
-        if self._on_live:
-            return seed(self._api)
-        live = self._api.db.model.get_connection()
-        if live.in_transaction:
-            raise RuntimeError("the live connection has an open transaction")
-        version = live.execute("PRAGMA data_version").fetchone()[0]
-        scratch = self._scratch_database()
-        target = self._scratch_connection(scratch)
-        with target.using():
-            live.backup(target)
-        before = table_digests(target)
-        target_api = _SeedApi(self._api, scratch)
-        result = seed(target_api)
-        target = self._scratch_connection(scratch)
-        if target.in_transaction:
-            raise RuntimeError("the seed left a scratch transaction open")
-        after = table_digests(target)
-        changed = [table for table in after if after[table] != before.get(table)]
-        if set(before) != set(after):
-            raise RuntimeError("the seed changed the database schema")
-        if live.execute("PRAGMA data_version").fetchone()[0] != version:
-            raise RuntimeError("the live database changed while the seed ran")
-        self._publish(live, scratch.model.db_path, changed, version)
-        for name, args, kwargs in target_api.deferred:
-            getattr(self._api, name)(*args, **kwargs)
-        return result
-
-    def _scratch_database(self):
-        if self._scratch is None:
-            from database.operations import PromptDatabase
-
-            self._directory = self._scratch_directory()
-            self._scratch = PromptDatabase(str(self._directory / "seed.db"))
-        return self._scratch
-
-    def _scratch_connection(self, scratch) -> sqlite3.Connection:
-        """The calling thread's scratch connection, set never to sync."""
-        conn = scratch.model.get_connection()
-        if conn not in self._unsynced:
-            conn.execute("PRAGMA synchronous = OFF")
-            self._unsynced.add(conn)
-        return conn
-
-    @staticmethod
-    def _publish(
-        live: sqlite3.Connection, path: str, tables: list[str], version: int
-    ) -> None:
-        """Replace ``tables`` with the scratch rows in one live transaction.
-
-        ``version`` is the live ``data_version`` read before the seed. It is
-        checked again under the write lock: a writer whose transaction was
-        open during the seed commits before ``BEGIN IMMEDIATE`` returns, and
-        its rows must fail the reset instead of being overwritten.
-        """
-        if not tables:
-            return
-        foreign_keys = live.execute("PRAGMA foreign_keys").fetchone()[0]
-        # Rows are copied as they are; no cascade may reach a table left alone.
-        live.execute("PRAGMA foreign_keys = OFF")
-        try:
-            live.execute(f"ATTACH DATABASE ? AS {SCRATCH_SCHEMA}", (path,))
-            try:
-                live.execute("BEGIN IMMEDIATE")
-                try:
-                    if live.execute("PRAGMA data_version").fetchone()[0] != version:
-                        raise RuntimeError(
-                            "the live database changed while the seed ran"
-                        )
-                    for table in tables:
-                        columns = ", ".join(
-                            _quote(c) if c != "rowid" else "rowid"
-                            for c in _columns(live, table)
-                        )
-                        name = _quote(table)
-                        live.execute(f"DELETE FROM main.{name}")
-                        live.execute(
-                            f"INSERT INTO main.{name} ({columns}) "
-                            f"SELECT {columns} FROM {SCRATCH_SCHEMA}.{name}"
-                        )
-                    live.commit()
-                except BaseException:
-                    live.rollback()
-                    raise
-            finally:
-                live.execute(f"DETACH DATABASE {SCRATCH_SCHEMA}")
-        finally:
-            live.execute(f"PRAGMA foreign_keys = {int(foreign_keys)}")
-
-    def close(self) -> None:
-        """Close every scratch connection, then remove the scratch files."""
-        if self._on_live:
-            return
-        if self._scratch is not None:
-            self._scratch.close_all()
-            self._scratch = None
-        if self._directory is not None:
-            shutil.rmtree(self._directory, ignore_errors=True)
-            self._directory = None
-
-    def _scratch_directory(self) -> Path:
-        """A folder in memory (tmpfs) where there is one, else ``directory``.
-
-        Even unsynced, scratch writes on the runner's disk wait whenever the
-        kernel is writing back the pages they rewrite.
-        """
-        root = self._memory_root
-        if root is not None and root.is_dir() and os.access(root, os.W_OK):
-            remove_orphaned_scratch(root)
-            if shutil.disk_usage(root).free >= MEMORY_FREE_BYTES:
-                prefix = f"{SCRATCH_PREFIX}{os.getpid()}-"
-                return Path(tempfile.mkdtemp(prefix=prefix, dir=root))
-        self._fallback.mkdir(parents=True, exist_ok=True)
-        return self._fallback

tests/browser/folder_trash.py

@@ -1,78 +0,0 @@
-"""Empty a browser-test media folder without waiting for the disk (#384).
-
-A reset used to ``shutil.rmtree`` the output and input folders the last
-fixture wrote. Unlinking a file whose pages are still being written back
-waits for that write, so on a slow CI disk a reset stalled in ``rmtree``.
-``replace_folder`` renames the old folder into ``<parent>/reset-trash`` (one
-metadata change) and makes a new empty one; a background thread deletes the
-trash, and ``close_trash`` stops that thread and removes what is left before
-the server's temp root goes.
-"""
-
-from __future__ import annotations
-
-import queue
-import shutil
-import threading
-import uuid
-from pathlib import Path
-
-TRASH_NAME = "reset-trash"
-
-
-class _Trash:
-    def __init__(self, directory: Path) -> None:
-        self.directory = directory
-        self._queue: "queue.SimpleQueue[Path | None]" = queue.SimpleQueue()
-        self._thread = threading.Thread(
-            target=self._work, name="pm-reset-trash", daemon=True
-        )
-        self._thread.start()
-
-    def _work(self) -> None:
-        while (folder := self._queue.get()) is not None:
-            shutil.rmtree(folder, ignore_errors=True)
-
-    def discard(self, folder: Path) -> None:
-        self.directory.mkdir(exist_ok=True)
-        moved = self.directory / f"{folder.name}-{uuid.uuid4().hex}"
-        folder.rename(moved)
-        self._queue.put(moved)
-
-    def close(self) -> None:
-        self._queue.put(None)
-        self._thread.join()
-        shutil.rmtree(self.directory, ignore_errors=True)
-
-
-_trashes: dict[Path, _Trash] = {}
-# Trash folders whose server has cleaned up: a reset still running then
-# deletes in place instead of starting a thread nobody would stop.
-_closed: set[Path] = set()
-_lock = threading.Lock()
-
-
-def replace_folder(folder: Path) -> None:
-    """Make ``folder`` an empty folder; its old content is deleted later."""
-    folder = Path(folder)
-    if folder.exists():
-        directory = folder.parent / TRASH_NAME
-        with _lock:
-            trash = _trashes.get(directory)
-            if trash is None and directory not in _closed:
-                trash = _trashes[directory] = _Trash(directory)
-            if trash is not None:
-                trash.discard(folder)
-        if folder.exists():
-            shutil.rmtree(folder)
-    folder.mkdir()
-
-
-def close_trash(parent: Path) -> None:
-    """Stop deleting under ``parent``'s trash and remove it (blocking)."""
-    directory = Path(parent) / TRASH_NAME
-    with _lock:
-        _closed.add(directory)
-        trash = _trashes.pop(directory, None)
-    if trash is not None:
-        trash.close()

tests/browser/reset_equivalence.py

@@ -1,279 +0,0 @@
-"""Prove the scratch-seeded resets leave the state the old resets left (#384).
-
-Run by hand on Linux (the browser server's platform):
-
-    python -m tests.browser.reset_equivalence
-
-It starts the browser server three times: twice seeding on the live database
-as before #384 (``PM_TEST_SEED_ON_LIVE=1``) and once with the scratch seeder.
-Each server gets the same sequence: every reset kind of ``/__test__/reset``
-and ``/__test__/concept/reset``, with other fixtures' resets and API writes
-between them. After every reset it reads every table's rows (by rowid), the
-output and input file trees (size, SHA-256, mtime) and the response body.
-
-Values that differ between the two old runs are wall-clock values (created_at,
-an unpinned concept "now", file mtimes); they are masked. Everything else must
-be equal between the old runs and the new one. Exit status 0 means equal.
-"""
-
-from __future__ import annotations
-
-import hashlib
-import json
-import os
-import sqlite3
-import subprocess
-import sys
-import tempfile
-import time
-import urllib.request
-from pathlib import Path
-
-REPO_ROOT = Path(__file__).resolve().parents[2]
-READY_TIMEOUT_S = 60
-CLOCK = "<clock>"
-
-RESETS = [
-    ("/__test__/reset", {}),
-    ("/__test__/reset", {"extra_images": 2}),
-    ("/__test__/reset", {"extra_images": 108}),
-    ("/__test__/reset", {"include_video": True}),
-    ("/__test__/reset", {"prompt_inspector": True}),
-    ("/__test__/reset", {"native_library": True}),
-    ("/__test__/reset", {"native_library": True, "extra_prompts": 125}),
-    ("/__test__/reset", {"native_library": True, "extra_variants": 41}),
-    ("/__test__/reset", {"workspace_visual": True}),
-    ("/__test__/reset", {"full_inspector": True}),
-    (
-        "/__test__/reset",
-        {"full_inspector": True, "include_video": True, "extra_images": 2},
-    ),
-    ("/__test__/reset", {"chronological_history": True}),
-    ("/__test__/reset", {"chronological_history": True, "history_runs": 85}),
-    ("/__test__/concept/reset", {}),
-    ("/__test__/concept/reset", {"extra_images": 100}),
-    ("/__test__/concept/reset", {"seedless_images": 2, "short_seedless": True}),
-    ("/__test__/concept/reset", {"now": 1_780_000_000}),
-]
-# Writes by other fixtures and the API between resets; the resets must keep or
-# replace them exactly as before.
-DIRTY = [
-    ("/__test__/tagging/reset", {}),
-    ("/__test__/system/reset", None),
-    (
-        "/__test__/saved-views/raw",
-        {"page": "images", "name": "dirty", "state": "{}"},
-    ),
-]
-
-
-class Server:
-    def __init__(self, on_live: bool) -> None:
-        self.tmp = tempfile.TemporaryDirectory(prefix="pm-equivalence-")
-        env = {**os.environ, "TMPDIR": self.tmp.name}
-        env.update(TEMP=self.tmp.name, TMP=self.tmp.name)
-        env.pop("PM_TEST_SEED_ON_LIVE", None)
-        if on_live:
-            env["PM_TEST_SEED_ON_LIVE"] = "1"
-        self.process = subprocess.Popen(
-            [
-                sys.executable,
-                "-m",
-                "tests.browser.server",
-                "--port",
-                "0",
-                "--exit-on-stdin-eof",
-            ],
-            cwd=REPO_ROOT,
-            env=env,
-            stdin=subprocess.PIPE,
-            stdout=subprocess.PIPE,
-            stderr=subprocess.DEVNULL,
-        )
-        line = self.process.stdout.readline().decode()
-        if not line.startswith("PM_TEST_SERVER_PORT="):
-            raise RuntimeError(f"no port line: {line!r}")
-        self.base = f"http://127.0.0.1:{int(line.split('=', 1)[1])}"
-        deadline = time.monotonic() + READY_TIMEOUT_S
-        while True:
-            try:
-                self.post_json("GET", "/prompt_manager/test", None)
-                break
-            except OSError:
-                if time.monotonic() > deadline or self.process.poll() is not None:
-                    raise
-                time.sleep(0.1)
-        (self.root,) = Path(self.tmp.name).glob("pm-browser-*")
-
-    def post_json(self, method: str, path: str, data) -> dict:
-        body = None if data is None else json.dumps(data).encode()
-        request = urllib.request.Request(
-            self.base + path,
-            data=body if method == "POST" else None,
-            method=method,
-            headers={"Content-Type": "application/json"} if body else {},
-        )
-        with urllib.request.urlopen(request, timeout=120) as response:  # nosec B310
-            text = response.read().decode()
-        try:
-            return json.loads(text)
-        except ValueError:
-            return {"text": text[:200]}
-
-    def stop(self) -> None:
-        self.process.stdin.close()
-        self.process.wait(timeout=READY_TIMEOUT_S)
-        self.process.stdout.close()
-        self.tmp.cleanup()
-
-
-def read_tables(path: Path) -> dict:
-    conn = sqlite3.connect(path.as_uri() + "?mode=ro", uri=True)
-    try:
-        tables = {}
-        names = [
-            row[0]
-            for row in conn.execute(
-                "SELECT name FROM sqlite_master WHERE type='table' ORDER BY name"
-            )
-        ]
-        for name in names:
-            quoted = '"' + name.replace('"', '""') + '"'
-            columns = [row[1] for row in conn.execute(f"PRAGMA table_info({quoted})")]
-            try:
-                cursor = conn.execute(f"SELECT rowid, * FROM {quoted} ORDER BY rowid")
-                columns = ["rowid", *columns]
-            except sqlite3.OperationalError:
-                cursor = conn.execute(f"SELECT * FROM {quoted} ORDER BY 1")
-            rows = [
-                [value.hex() if isinstance(value, bytes) else value for value in row]
-                for row in cursor
-            ]
-            tables[name] = {"columns": columns, "rows": rows}
-        return tables
-    finally:
-        conn.close()
-
-
-def read_files(root: Path) -> dict:
-    files = {}
-    for top in ("output", "input"):
-        for path in sorted((root / top).rglob("*")):
-            if path.is_file():
-                stat = path.stat()
-                files[path.relative_to(root).as_posix()] = [
-                    stat.st_size,
-                    hashlib.sha256(path.read_bytes()).hexdigest(),
-                    stat.st_mtime_ns,
-                ]
-    return files
-
-
-def flatten(value, prefix="") -> dict:
-    if isinstance(value, dict):
-        out = {}
-        for key, item in value.items():
-            out.update(flatten(item, f"{prefix}/{key}"))
-        return out
-    if isinstance(value, list):
-        out = {f"{prefix}#len": len(value)}
-        for index, item in enumerate(value):
-            out.update(flatten(item, f"{prefix}[{index}]"))
-        return out
-    return {prefix: value}
-
-
-def snapshot(server: Server, response: dict) -> dict:
-    flat = {}
-    for table, data in read_tables(server.root / "browser.db").items():
-        flat[f"table {table}#rows"] = len(data["rows"])
-        for index, row in enumerate(data["rows"]):
-            for column, value in zip(data["columns"], row):
-                flat[f"table {table}[{index}].{column}"] = value
-    for name, (size, digest, mtime) in read_files(server.root).items():
-        flat[f"file {name}"] = [size, digest]
-        flat[f"file {name}#mtime"] = mtime
-    flat["files#count"] = sum(
-        1 for key in flat if key.startswith("file ") and "#" not in key
-    )
-    for key, value in flatten(response).items():
-        flat[f"response{key}"] = value
-    # Each server has its own temp root and a random storage epoch; name them
-    # instead of masking every column that carries them.
-    epoch = flat.get("table execution_history_metadata[0].storage_epoch")
-    names = [(str(server.root), "<root>")]
-    if isinstance(epoch, str) and epoch:
-        names.append((epoch, "<epoch>"))
-    for key, value in flat.items():
-        if isinstance(value, str):
-            for text, name in names:
-                value = value.replace(text, name)
-            flat[key] = value
-    return flat
-
-
-def run(on_live: bool) -> list[dict]:
-    server = Server(on_live)
-    try:
-        states = []
-        for index, (path, data) in enumerate(RESETS * 2):
-            dirty_path, dirty_data = DIRTY[index % len(DIRTY)]
-            server.post_json("POST", dirty_path, dirty_data)
-            started = time.monotonic()
-            response = server.post_json("POST", path, data)
-            elapsed = time.monotonic() - started
-            print(f"  {'old' if on_live else 'new'} {path} {data}: {elapsed:.2f}s")
-            states.append(snapshot(server, response))
-        return states
-    finally:
-        server.stop()
-
-
-def column_key(key: str) -> str:
-    """``table t[3].created_at`` -> ``table t.created_at`` (mask whole columns)."""
-    if key.startswith("table ") and "[" in key:
-        head, _, tail = key.partition("[")
-        return head + tail.partition("]")[2]
-    return key
-
-
-def main() -> int:
-    old_a, old_b, new = run(True), run(True), run(False)
-    failures = 0
-    masked_all = set()
-    for step, (path, data) in enumerate(RESETS * 2):
-        a, b, n = old_a[step], old_b[step], new[step]
-        masked = {
-            column_key(key)
-            for key in set(a) | set(b)
-            if a.get(key, CLOCK) != b.get(key, CLOCK)
-        }
-        masked_all |= masked
-        keys = set(a) | set(n)
-        diffs = [
-            key
-            for key in sorted(keys)
-            if column_key(key) not in masked
-            and a.get(key, "<missing>") != n.get(key, "<missing>")
-        ]
-        tables = len(
-            {k.split("[")[0].split("#")[0] for k in a if k.startswith("table ")}
-        )
-        status = "EQUAL" if not diffs else f"DIFFERENT ({len(diffs)} values)"
-        print(
-            f"{status}: {path} {json.dumps(data)}: {tables} tables, "
-            f"{a['files#count']} files, {len(keys)} values, "
-            f"{len(masked)} masked columns"
-        )
-        for key in diffs[:20]:
-            old, now = a.get(key, "<missing>"), n.get(key, "<missing>")
-            print(f"    {key}: old {old!r} new {now!r}")
-        failures += bool(diffs)
-    print("masked (differ between the two old runs):")
-    for key in sorted(masked_all):
-        print(f"    {key}")
-    return 1 if failures else 0
-
-
-if __name__ == "__main__":
-    sys.exit(main())

tests/browser/server.py

@@ -19,10 +19,8 @@ from aiohttp.web_runner import GracefulExit
 from PIL import Image
 from PIL.PngImagePlugin import PngInfo
 
-from tests.browser.folder_trash import close_trash, replace_folder
 from tests.browser.request_hold import register_request_holds
 from tests.browser import slow_reset_watch
-from tests.browser.fast_reset import SEED_ON_LIVE_ENV, ScratchSeeder
 from tests.browser.workspace_activity_fixture import register_activity_commands
 from tests.browser.workspace_system_fixture import register_system_commands
 from tests.browser.workspace_integrations_fixture import (
@@ -227,10 +225,6 @@ def create_app(root: Path) -> web.Application:
     register_integrations_commands(app, routes, api, root)
     request_holds = register_request_holds(app, routes)
     slow_reset_watch.install(app, api)
-    # Resets seed a scratch copy and publish it in one commit (#384).
-    seeder = ScratchSeeder(
-        api, root / "reset-scratch", on_live=os.environ.get(SEED_ON_LIVE_ENV) == "1"
-    )
 
     async def canonical_asset(request: web.Request) -> web.Response:
         filename, content_type = CANONICAL_ASSETS[request.path]
@@ -244,7 +238,6 @@ def create_app(root: Path) -> web.Application:
         routes.get(path)(canonical_asset)
 
     def seed(
-        target,
         extra_images: int = 0,
         include_video: bool = False,
         prompt_inspector: bool = False,
@@ -252,7 +245,7 @@ def create_app(root: Path) -> web.Application:
         extra_prompts: int = 0,
         extra_variants: int = 0,
     ) -> None:
-        with target.db.model.get_connection() as conn:
+        with api.db.model.get_connection() as conn:
             for table in (
                 "saved_views",
                 "image_favorites",
@@ -271,7 +264,8 @@ def create_app(root: Path) -> web.Application:
             ):
                 conn.execute(f"DELETE FROM {table}")
         for name in ("output", "input"):
-            replace_folder(root / name)
+            shutil.rmtree(root / name)
+            (root / name).mkdir()
         for index, (directory, text, tag) in enumerate(
             [
                 ("output", "Cat portrait", "cats"),
@@ -314,30 +308,28 @@ def create_app(root: Path) -> web.Application:
                     image.save(image_path, pnginfo=info)
             stamp = time.time() - index * 10
             os.utime(image_path, (stamp, stamp))
-            pid = target.db.save_prompt(
+            pid = api.db.save_prompt(
                 text, tags=[tag], prompt_hash=generate_prompt_hash(text)
             )
-            target.db.link_image_to_prompt(pid, str(image_path))
+            api.db.link_image_to_prompt(pid, str(image_path))
             if index == 0:
                 video_prompt_id = pid
                 for number in range(extra_images):
                     extra = root / directory / "bulk" / f"{number:04}.png"
                     extra.parent.mkdir(exist_ok=True)
                     shutil.copyfile(image_path, extra)
-                    target.db.link_image_to_prompt(pid, str(extra))
+                    api.db.link_image_to_prompt(pid, str(extra))
         if native_library:
-            positive = target.db.get_prompt_by_hash(
-                generate_prompt_hash("Cat portrait")
-            )
+            positive = api.db.get_prompt_by_hash(generate_prompt_hash("Cat portrait"))
             positive_id = positive["id"]
-            target.db.update_prompt_metadata(
+            api.db.update_prompt_metadata(
                 positive_id, category="Animals", tags=["cats", "studio"]
             )
             variants = []
             for index, negative in enumerate(
                 ["blurry, watermark", "low quality", "low contrast", ""]
             ):
-                pair_id = target.db.save_prompt_pair(
+                pair_id = api.db.save_prompt_pair(
                     "Cat portrait",
                     negative,
                     title=f"Portrait variant {index + 1}",
@@ -345,30 +337,28 @@ def create_app(root: Path) -> web.Application:
                     tags=["studio"] if index == 0 else ["outdoors"],
                 )
                 variants.append(pair_id)
-            with target.db.model.get_connection() as conn:
+            with api.db.model.get_connection() as conn:
                 images = conn.execute(
                     "SELECT id FROM generated_images WHERE prompt_id=? ORDER BY id",
                     (positive_id,),
                 ).fetchall()
             for index, image in enumerate(images[:2]):
-                target.db.link_image_to_pair(
+                api.db.link_image_to_pair(
                     variants[index], image["id"], {"source": "fixture", "anchor": index}
                 )
-            target.db.link_image_to_pair(
+            api.db.link_image_to_pair(
                 variants[0], images[0]["id"], {"source": "fixture second evidence"}
             )
-            target.db.record_prompt_pair_use(variants[0], times=3)
+            api.db.record_prompt_pair_use(variants[0], times=3)
             for number in range(extra_variants):
-                target.db.save_prompt_pair(
-                    "Cat portrait", f"Negative extra {number:04}"
-                )
+                api.db.save_prompt_pair("Cat portrait", f"Negative extra {number:04}")
             legacy = root / "output" / "legacy-unassigned.png"
             shutil.copyfile(root / "output" / "shared" / "same.png", legacy)
-            target.db.link_image_to_prompt(positive_id, str(legacy))
-            target.db.save_prompt("Dog landscape", category="Animals", tags=["dogs"])
-            target.db.save_prompt_pair("Dual role portrait", "Cat portrait")
+            api.db.link_image_to_prompt(positive_id, str(legacy))
+            api.db.save_prompt("Dog landscape", category="Animals", tags=["dogs"])
+            api.db.save_prompt_pair("Dual role portrait", "Cat portrait")
             for number in range(extra_prompts):
-                target.db.save_prompt(
+                api.db.save_prompt(
                     f"Library extra {number:04}", category="Bulk", tags=["bulk"]
                 )
         if prompt_inspector:
@@ -389,8 +379,8 @@ def create_app(root: Path) -> web.Application:
         from py.api.thumbnail_settings import ThumbnailSettings
 
         # 150 px previews in the default format (#317 thumbnail settings).
-        target._generate_gallery_thumbnails_sync(
-            target._get_all_output_dirs(), ThumbnailSettings("webp", 80, 150)
+        api._generate_gallery_thumbnails_sync(
+            api._get_all_output_dirs(), ThumbnailSettings("webp", 80, 150)
         )
         if include_video:
             # Seed after thumbnail generation to exercise a real unthumbnailed video.
@@ -398,8 +388,8 @@ def create_app(root: Path) -> web.Application:
             shutil.copyfile(
                 Path(__file__).resolve().parents[1] / "fixtures" / "tiny.webm", video
             )
-            target.db.link_image_to_prompt(video_prompt_id, str(video))
-        target.invalidate_gallery_cache()
+            api.db.link_image_to_prompt(video_prompt_id, str(video))
+        api.invalidate_gallery_cache()
 
     history_fixture = None
 
@@ -444,10 +434,7 @@ def create_app(root: Path) -> web.Application:
             raise web.HTTPBadRequest()
         if chronological_history:
             manifest = await api._run_in_executor(
-                seeder.run,
-                lambda target: seed_workspace_history(
-                    target, root, history_runs=history_runs
-                ),
+                lambda: seed_workspace_history(api, root, history_runs=history_runs)
             )
             history_fixture = manifest
             return web.json_response({"success": True, **manifest})
@@ -505,31 +492,25 @@ def create_app(root: Path) -> web.Application:
             raise web.HTTPBadRequest()
         if workspace_visual:
             history_fixture = None
-            await api._run_in_executor(
-                seeder.run, lambda target: seed_workspace_visual(target, root)
-            )
+            await api._run_in_executor(seed_workspace_visual, api, root)
             return web.json_response({"success": True})
         if full_inspector:
             history_fixture = None
             await api._run_in_executor(
-                seeder.run,
-                lambda target: seed_workspace_inspector(
-                    target, root, extra_images=extra, include_video=include_video
-                ),
+                lambda: seed_workspace_inspector(
+                    api, root, extra_images=extra, include_video=include_video
+                )
             )
             return web.json_response({"success": True})
         history_fixture = None
         await api._run_in_executor(
-            seeder.run,
-            lambda target: seed(
-                target,
-                extra,
-                include_video,
-                prompt_inspector,
-                native_library,
-                extra_prompts,
-                extra_variants,
-            ),
+            seed,
+            extra,
+            include_video,
+            prompt_inspector,
+            native_library,
+            extra_prompts,
+            extra_variants,
         )
         return web.json_response({"success": True})
 
@@ -569,15 +550,14 @@ def create_app(root: Path) -> web.Application:
         reveal_stub.reset()
         history_fixture = None
         manifest = await api._run_in_executor(
-            seeder.run,
-            lambda target: seed_workspace_concept(
-                target,
+            lambda: seed_workspace_concept(
+                api,
                 root,
                 extra_images=data.get("extra_images", 0),
                 seedless_images=data.get("seedless_images", 0),
                 short_seedless=data.get("short_seedless", False),
                 now=data.get("now"),
-            ),
+            )
         )
         return web.json_response({"success": True, **manifest})
 
@@ -645,12 +625,9 @@ def create_app(root: Path) -> web.Application:
     async def cleanup(application: web.Application) -> None:
         # A scan of an ambiguous fixture file captures it for Needs review.
         api.close_annotation_captures()
-        seeder.close()
         api.db.close_all()
-        # Old media folders still waiting for deletion go before the temp root.
-        close_trash(root)
 
-    seed(api)
+    seed()
     # Open Needs review change streams end first, as on the real host.
     app.on_shutdown.append(api.close_review_events)
     app.on_cleanup.append(cleanup)

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/browser/workspace_concept_fixture.py

@@ -16,6 +16,7 @@ import json
 import math
 import os
 from pathlib import Path
+import shutil
 import time
 from types import SimpleNamespace
 from typing import TYPE_CHECKING
@@ -33,7 +34,6 @@ from utils.execution_contexts import (
 from utils.execution_history_events import make_finish, make_start
 from utils.execution_outputs import OutputBundle, OutputReference, snapshot_output_roots
 from utils.pair_execution import EXECUTION_KEY
-from tests.browser.folder_trash import replace_folder
 
 if TYPE_CHECKING:
     from py.api import PromptManagerAPI
@@ -300,7 +300,8 @@ def _clear(api: PromptManagerAPI, root: Path) -> None:
         for table in _TABLES:
             conn.execute(f"DELETE FROM {table}")
     for name in ("output", "input"):
-        replace_folder(root / name)
+        shutil.rmtree(root / name)
+        (root / name).mkdir()
 
 
 def _record_run(

tests/browser/workspace_library_editor_fixture.py

@@ -3,11 +3,11 @@
 from __future__ import annotations
 
 from pathlib import Path
+import shutil
 from typing import TYPE_CHECKING
 
 from aiohttp import web
 from PIL import Image
-from tests.browser.folder_trash import replace_folder
 
 if TYPE_CHECKING:
     from py.api import PromptManagerAPI
@@ -51,7 +51,8 @@ def seed_library_editor(api: PromptManagerAPI, root: Path) -> dict:
         for table in _TABLES:
             conn.execute(f"DELETE FROM {table}")
     for name in ("output", "input"):
-        replace_folder(root / name)
+        shutil.rmtree(root / name)
+        (root / name).mkdir()
     epoch = db.get_execution_epoch()
     receipt = db.create_library_recipe(
         EDITOR_POSITIVE,

tests/browser/workspace_review_fixture.py

@@ -52,7 +52,6 @@ from typing import TYPE_CHECKING
 from aiohttp import web
 from PIL import Image
 from PIL.PngImagePlugin import PngInfo
-from tests.browser.folder_trash import replace_folder
 
 if TYPE_CHECKING:
     from py.api import PromptManagerAPI
@@ -226,7 +225,8 @@ def seed_review(
             conn.execute(f"DELETE FROM {table}")
     api.close_annotation_captures()
     for name in ("output", "input"):
-        replace_folder(root / name)
+        shutil.rmtree(root / name)
+        (root / name).mkdir()
     shutil.rmtree(root / "temp" / "review", ignore_errors=True)
     output = root / "output"
     files = []

tests/browser/workspace_tagging_fixture.py

@@ -19,7 +19,6 @@ from typing import TYPE_CHECKING
 from aiohttp import web
 from PIL import Image
 from PIL.PngImagePlugin import PngInfo
-from tests.browser.folder_trash import replace_folder
 
 if TYPE_CHECKING:
     from py.api import PromptManagerAPI
@@ -99,7 +98,8 @@ def seed_tagging(api: PromptManagerAPI, root: Path, extra_images: int = 0) -> di
         for table in _TABLES:
             conn.execute(f"DELETE FROM {table}")
     for name in ("output", "input"):
-        replace_folder(root / name)
+        shutil.rmtree(root / name)
+        (root / name).mkdir()
     output = root / "output"
     _png(output / FILES[0][0], (196, 140, 92))
     _png(output / FILES[1][0], (150, 110, 80))

tests/browser/workspace_visual_fixture.py

@@ -5,13 +5,13 @@ from __future__ import annotations
 import json
 import os
 from pathlib import Path
+import shutil
 from typing import TYPE_CHECKING
 
 from PIL import Image, ImageDraw
 from PIL.PngImagePlugin import PngInfo
 
 from utils.hashing import generate_prompt_hash
-from tests.browser.folder_trash import replace_folder
 
 if TYPE_CHECKING:
     from py.api import PromptManagerAPI
@@ -129,7 +129,8 @@ def seed_workspace_visual(api: PromptManagerAPI, root: Path) -> None:
         ):
             conn.execute(f"DELETE FROM {table}")
     for name in ("output", "input"):
-        replace_folder(root / name)
+        shutil.rmtree(root / name)
+        (root / name).mkdir()
     for parent_index, (positive, negatives) in enumerate(zip(_POSITIVES, _NEGATIVES)):
         category = ("Architecture", "Landscape", "Abstract")[parent_index % 3]
         tags = ["geometric", ("warm", "cool")[parent_index % 2]]

tests/browser/workspace_vocabulary_fixture.py

@@ -7,10 +7,10 @@ in concept order, so tag creation order and category first use match the mock.
 from __future__ import annotations
 
 from pathlib import Path
+import shutil
 from typing import TYPE_CHECKING
 
 from aiohttp import web
-from tests.browser.folder_trash import replace_folder
 
 if TYPE_CHECKING:
     from py.api import PromptManagerAPI
@@ -120,7 +120,8 @@ def seed_vocabulary(
         for table in _TABLES:
             conn.execute(f"DELETE FROM {table}")
     for name in ("output", "input"):
-        replace_folder(root / name)
+        shutil.rmtree(root / name)
+        (root / name).mkdir()
     epoch = db.get_execution_epoch()
     recipes = []
     for recipe in CONCEPT_RECIPES:

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

@@ -1,409 +0,0 @@
-"""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

@@ -1,419 +0,0 @@
-"""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,10 +5,9 @@ import os
 import shutil
 import sys
 import tempfile
-import threading
 import unittest
 from pathlib import Path
-from unittest.mock import MagicMock, patch
+from unittest.mock import MagicMock
 
 _mock_server = MagicMock()
 _mock_server.PromptServer.instance.routes = MagicMock()
@@ -21,9 +20,7 @@ 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:
@@ -41,16 +38,11 @@ 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)
 
@@ -62,36 +54,23 @@ class FakeLoggerManager:
 
     def get_log_files(self):
         files = []
-        for row in sorted(
-            logging_config.scan_log_folder(self.log_dir), key=lambda r: r["name"]
-        ):
-            if row["_dir"]:
-                continue
+        for path in sorted(self.log_dir.glob("*.log*")):
             files.append(
                 {
-                    "filename": row["name"],
-                    "path": str(self.log_dir / row["name"]),
-                    "size": row["size"],
+                    "filename": path.name,
+                    "path": str(path),
+                    "size": path.stat().st_size,
                     "modified": "2026-10-07T00:00:00",
-                    "is_main": row["kind"] == "active",
-                    "kind": row["kind"],
+                    "is_main": path.name == "prompt_manager.log",
                 }
             )
         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 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"])
+        for path in self.log_dir.glob("prompt_manager.log*"):
+            path.write_text("")
+            truncated.append(path.name)
         return {"truncated": truncated, "errors": []}
 
     def get_log_stats(self):
@@ -399,129 +378,12 @@ class TestLogFilesAndDownload(LoggingAPITestCase):
         self.assertEqual(resp.status, 400)
 
     async def test_download_read_failure_is_500(self):
-        # 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"
-            )
+        (self.log_dir / "dir.log").mkdir()
+        resp = await self.client.request("GET", "/prompt_manager/logs/download/dir.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):
 
@@ -533,40 +395,6 @@ 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_browser_fast_reset.py

@@ -1,335 +0,0 @@
-"""Browser-test resets seed a scratch copy and publish it in one commit (#384)."""
-
-import hashlib
-import os
-import sqlite3
-import subprocess
-import sys
-import tempfile
-import threading
-import time
-import types
-import unittest
-from pathlib import Path
-from unittest import mock
-
-try:
-    from tests.open_handles import assert_closed
-except ImportError:  # discovered with tests/ as the top-level directory
-    from open_handles import assert_closed
-
-from database.operations import PromptDatabase
-from tests.browser import fast_reset
-from tests.browser.fast_reset import ScratchSeeder, table_digests
-
-
-def digests(db: PromptDatabase) -> dict:
-    return table_digests(db.model.get_connection())
-
-
-def add_prompt(db: PromptDatabase, text: str) -> int:
-    with db.model.get_connection() as conn:
-        return conn.execute(
-            "INSERT INTO prompts (text, hash, created_at, updated_at) "
-            "VALUES (?, ?, '2026-01-01 00:00:00', '2026-01-01 00:00:00')",
-            (text, hashlib.sha256(text.encode()).hexdigest()),
-        ).lastrowid
-
-
-def add_image(db: PromptDatabase, prompt_id: int, name: str) -> int:
-    with db.model.get_connection() as conn:
-        return conn.execute(
-            "INSERT INTO generated_images (prompt_id, image_path, filename, "
-            "generation_time) VALUES (?, ?, ?, '2026-01-01 00:00:00')",
-            (prompt_id, f"/out/{name}", name),
-        ).lastrowid
-
-
-def reseed(api) -> str:
-    """A seeder in the old style: clears its tables, then many small commits."""
-    db = api.db
-    with db.model.get_connection() as conn:
-        for table in ("generated_images", "prompt_tags", "tags", "prompts"):
-            conn.execute(f"DELETE FROM {table}")
-    for index in range(30):
-        prompt_id = add_prompt(db, f"seeded {index}")
-        add_image(db, prompt_id, f"seeded-{index}.png")
-        with db.model.get_connection() as conn:
-            tag_id = conn.execute(
-                "INSERT INTO tags (name) VALUES (?)", (f"tag{index}",)
-            ).lastrowid
-            conn.execute(
-                "INSERT INTO prompt_tags (prompt_id, tag_id) VALUES (?, ?)",
-                (prompt_id, tag_id),
-            )
-    return "manifest"
-
-
-class TestScratchSeeder(unittest.TestCase):
-    def setUp(self):
-        self.tmp = tempfile.TemporaryDirectory()
-        self.addCleanup(self.tmp.cleanup)
-        self.root = Path(self.tmp.name)
-        self.addCleanup(assert_closed, self, self.tmp.name)
-        self.live = self.database("live.db")
-        self.api = types.SimpleNamespace(db=self.live, marker="live api")
-        self.seeder = ScratchSeeder(self.api, self.root / "scratch", memory_root=None)
-        self.addCleanup(self.seeder.close)
-        # Rows from "earlier tests": AUTOINCREMENT positions have moved on.
-        for index in range(5):
-            add_image(self.live, add_prompt(self.live, f"old {index}"), f"o{index}")
-        with self.live.model.get_connection() as conn:
-            conn.execute("DELETE FROM prompts WHERE text = 'old 4'")
-
-    def database(self, name: str) -> PromptDatabase:
-        db = PromptDatabase(str(self.root / name))
-        self.addCleanup(db.close_all)
-        return db
-
-    def oracle(self, seed):
-        """Run ``seed`` the old way, on a copy of the live database."""
-        path = self.root / "oracle.db"
-        target = sqlite3.connect(path)
-        try:
-            self.live.model.get_connection().backup(target)
-        finally:
-            target.close()
-        db = self.database("oracle.db")
-        seed(types.SimpleNamespace(db=db))
-        return digests(db)
-
-    def test_publishes_exactly_what_the_seed_writes_on_live(self):
-        expected = self.oracle(reseed)
-        self.assertEqual(self.seeder.run(reseed), "manifest")
-        self.assertEqual(digests(self.live), expected)
-        with self.live.model.get_connection() as conn:
-            first = conn.execute("SELECT MIN(id) FROM prompts").fetchone()[0]
-        self.assertEqual(first, 6, "AUTOINCREMENT continues as on live")
-
-    def test_runs_twice_from_the_state_the_first_run_left(self):
-        self.seeder.run(reseed)
-        expected = self.oracle(reseed)
-        self.seeder.run(reseed)
-        self.assertEqual(digests(self.live), expected)
-
-    def test_tables_the_seed_did_not_change_keep_their_rows(self):
-        before = digests(self.live)
-
-        def rename(api):
-            with api.db.model.get_connection() as conn:
-                conn.execute("UPDATE prompts SET text = 'renamed' WHERE id = 1")
-
-        self.seeder.run(rename)
-        after = digests(self.live)
-        changed = {table for table in after if after[table] != before[table]}
-        self.assertEqual(changed, {"prompts"})
-        with self.live.model.get_connection() as conn:
-            images = conn.execute("SELECT COUNT(*) FROM generated_images").fetchone()
-        # Replacing prompts' rows never cascades into generated_images.
-        self.assertEqual(images[0], 4)
-
-    def test_the_seed_sees_the_scratch_database_and_the_live_api(self):
-        seen = {}
-
-        def seed(api):
-            seen["db"] = api.db
-            seen["marker"] = api.marker
-            api.note = "set by the seed"
-
-        self.seeder.run(seed)
-        self.assertIsNot(seen["db"], self.live)
-        self.assertNotEqual(seen["db"].model.db_path, self.live.model.db_path)
-        self.assertEqual(seen["marker"], "live api")
-        self.assertEqual(self.api.note, "set by the seed")
-        self.assertIs(self.api.db, self.live)
-
-    def test_cache_drops_wait_until_the_rows_are_published(self):
-        seen = []
-
-        def invalidate_gallery_cache(retire=False):
-            with self.live.model.get_connection() as conn:
-                count = conn.execute("SELECT COUNT(*) FROM tags").fetchone()[0]
-            seen.append((count, retire))
-
-        self.api.invalidate_gallery_cache = invalidate_gallery_cache
-
-        def seed(api):
-            reseed(api)
-            api.invalidate_gallery_cache(retire=True)
-            self.assertEqual(seen, [], "not dropped while live has the old rows")
-
-        self.seeder.run(seed)
-        self.assertEqual(seen, [(30, True)])
-
-    def test_a_failing_seed_drops_no_cache(self):
-        seen = []
-        self.api.invalidate_gallery_cache = lambda: seen.append(True)
-
-        def broken(api):
-            api.invalidate_gallery_cache()
-            raise ValueError("seed failed")
-
-        with self.assertRaises(ValueError):
-            self.seeder.run(broken)
-        self.assertEqual(seen, [])
-
-    def test_a_failing_seed_leaves_live_untouched(self):
-        before = digests(self.live)
-
-        def broken(api):
-            reseed(api)
-            raise ValueError("seed failed")
-
-        with self.assertRaisesRegex(ValueError, "seed failed"):
-            self.seeder.run(broken)
-        self.assertEqual(digests(self.live), before)
-        self.assertIs(self.api.db, self.live)
-
-    def test_a_concurrent_live_commit_fails_the_reset(self):
-        other = self.database("live.db")
-
-        def racing(api):
-            reseed(api)
-            add_prompt(other, "written to live meanwhile")
-
-        with self.assertRaisesRegex(RuntimeError, "changed while"):
-            self.seeder.run(racing)
-        with self.live.model.get_connection() as conn:
-            texts = [row[0] for row in conn.execute("SELECT text FROM prompts")]
-        self.assertIn("written to live meanwhile", texts)
-        self.assertNotIn("seeded 0", texts)
-
-    def test_a_write_open_during_the_seed_and_committed_later_fails_it(self):
-        other = self.database("live.db")
-        began, finished = threading.Event(), []
-
-        def write():
-            with other.model.get_connection() as conn:
-                conn.execute("BEGIN IMMEDIATE")
-                conn.execute(
-                    "INSERT INTO prompts (text, hash) VALUES ('concurrent', 'c')"
-                )
-                began.set()
-                time.sleep(0.5)  # the publish waits on this lock meanwhile
-            other.model.close()
-            finished.append(True)
-
-        writer = threading.Thread(target=write)
-
-        def racing(api):
-            reseed(api)
-            writer.start()
-            self.assertTrue(began.wait(10))
-
-        with self.assertRaisesRegex(RuntimeError, "changed while"):
-            self.seeder.run(racing)
-        writer.join()
-        self.assertEqual(finished, [True])
-        with self.live.model.get_connection() as conn:
-            texts = [row[0] for row in conn.execute("SELECT text FROM prompts")]
-        self.assertIn("concurrent", texts)
-        self.assertNotIn("seeded 0", texts)
-        self.assert_live_connection_restored()
-
-    def test_a_failed_publish_leaves_live_and_its_connection_as_they_were(self):
-        before = digests(self.live)
-        columns = fast_reset._columns
-
-        def broken_columns(conn, table, *schema):
-            # The publish asks for the live table's columns without a schema.
-            return columns(conn, table, *schema) if schema else ["no_such_column"]
-
-        with mock.patch.object(fast_reset, "_columns", broken_columns):
-            with self.assertRaises(sqlite3.OperationalError):
-                self.seeder.run(reseed)
-        self.assertEqual(digests(self.live), before)
-        self.assert_live_connection_restored()
-
-    def assert_live_connection_restored(self):
-        conn = self.live.model.get_connection()
-        self.assertFalse(conn.in_transaction)
-        self.assertEqual(conn.execute("PRAGMA foreign_keys").fetchone()[0], 1)
-        names = [row[1] for row in conn.execute("PRAGMA database_list")]
-        self.assertEqual(names, ["main"])
-
-    def test_the_live_connection_is_left_as_it_was(self):
-        self.seeder.run(reseed)
-        conn = self.live.model.get_connection()
-        self.assertFalse(conn.in_transaction)
-        self.assertEqual(conn.execute("PRAGMA foreign_keys").fetchone()[0], 1)
-        names = [row[1] for row in conn.execute("PRAGMA database_list")]
-        self.assertEqual(names, ["main"])
-        self.assertEqual(conn.execute("PRAGMA foreign_key_check").fetchall(), [])
-
-    def test_on_live_mode_runs_the_seed_on_the_live_api(self):
-        seeder = ScratchSeeder(
-            self.api, self.root / "unused", on_live=True, memory_root=self.root
-        )
-        self.addCleanup(seeder.close)
-        expected = self.oracle(reseed)
-        seen = []
-        self.assertEqual(
-            seeder.run(lambda api: seen.append(api) or reseed(api)), "manifest"
-        )
-        self.assertIs(seen[0], self.api)
-        self.assertEqual(digests(self.live), expected)
-        self.assertFalse((self.root / "unused").exists())
-
-    def test_the_scratch_lives_under_the_memory_root_when_there_is_one(self):
-        memory = self.root / "memory"
-        memory.mkdir()
-        seeder = ScratchSeeder(self.api, self.root / "fallback", memory_root=memory)
-        self.addCleanup(seeder.close)
-        expected = self.oracle(reseed)
-        seeder.run(reseed)
-        self.assertEqual(digests(self.live), expected)
-        self.assertEqual(len(list(memory.glob("pm-reset-*/seed.db"))), 1)
-        self.assertFalse((self.root / "fallback").exists())
-        seeder.close()
-        self.assertEqual(list(memory.iterdir()), [])
-
-    def test_a_full_memory_root_puts_the_scratch_on_disk(self):
-        memory = self.root / "memory"
-        memory.mkdir()
-        seeder = ScratchSeeder(self.api, self.root / "fallback", memory_root=memory)
-        self.addCleanup(seeder.close)
-        full = mock.patch.object(
-            fast_reset.shutil,
-            "disk_usage",
-            return_value=types.SimpleNamespace(free=fast_reset.MEMORY_FREE_BYTES - 1),
-        )
-        with full:
-            seeder.run(reseed)
-        self.assertEqual(list(memory.iterdir()), [])
-        self.assertTrue((self.root / "fallback" / "seed.db").exists())
-
-    @unittest.skipUnless(os.name == "posix", "process probe is POSIX only")
-    def test_scratch_folders_of_dead_servers_are_removed(self):
-        memory = self.root / "memory"
-        memory.mkdir()
-        dead = subprocess.Popen([sys.executable, "-c", "pass"])
-        dead.wait()
-        orphan = memory / f"pm-reset-{dead.pid}-abc"
-        alive = memory / f"pm-reset-{os.getpid()}-def"
-        other = memory / "pm-reset-notapid"
-        for folder in (orphan, alive, other):
-            folder.mkdir()
-            (folder / "seed.db").write_bytes(b"x")
-        seeder = ScratchSeeder(self.api, self.root / "fallback", memory_root=memory)
-        self.addCleanup(seeder.close)
-        seeder.run(reseed)
-        self.assertFalse(orphan.exists())
-        self.assertTrue(alive.exists())
-        self.assertTrue(other.exists())
-        own = [p for p in memory.glob(f"pm-reset-{os.getpid()}-*") if p != alive]
-        self.assertEqual(len(own), 1)
-
-    def test_close_releases_the_scratch_files(self):
-        self.seeder.run(reseed)
-        self.seeder.close()
-        assert_closed(self, self.root / "scratch")
-        self.seeder.close()  # idempotent
-
-
-if __name__ == "__main__":
-    unittest.main()

tests/test_browser_folder_trash.py

@@ -1,67 +0,0 @@
-"""Browser-test media folders are emptied by a rename, deleted later (#384)."""
-
-import tempfile
-import threading
-import unittest
-from pathlib import Path
-
-from tests.browser.folder_trash import TRASH_NAME, close_trash, replace_folder
-
-
-class TestFolderTrash(unittest.TestCase):
-    def setUp(self):
-        self.tmp = tempfile.TemporaryDirectory()
-        self.addCleanup(self.tmp.cleanup)
-        self.root = Path(self.tmp.name)
-        self.addCleanup(close_trash, self.root)
-
-    def trash_threads(self):
-        return [t for t in threading.enumerate() if t.name == "pm-reset-trash"]
-
-    def test_replaces_the_folder_with_an_empty_one(self):
-        folder = self.root / "output"
-        (folder / "nested").mkdir(parents=True)
-        (folder / "nested" / "image.png").write_bytes(b"png")
-        replace_folder(folder)
-        self.assertTrue(folder.is_dir())
-        self.assertEqual(list(folder.iterdir()), [])
-        close_trash(self.root)
-        self.assertFalse((self.root / TRASH_NAME).exists())
-        self.assertEqual(self.trash_threads(), [])
-
-    def test_a_missing_folder_is_created_without_a_trash(self):
-        replace_folder(self.root / "input")
-        self.assertTrue((self.root / "input").is_dir())
-        self.assertFalse((self.root / TRASH_NAME).exists())
-        self.assertEqual(self.trash_threads(), [])
-
-    def test_repeated_replacements_share_one_thread_and_all_are_removed(self):
-        folder = self.root / "output"
-        for number in range(5):
-            folder.mkdir(exist_ok=True)
-            (folder / f"{number}.png").write_bytes(b"png")
-            replace_folder(folder)
-        self.assertEqual(len(self.trash_threads()), 1)
-        close_trash(self.root)
-        self.assertEqual(list(folder.iterdir()), [])
-        self.assertFalse((self.root / TRASH_NAME).exists())
-        close_trash(self.root)  # idempotent
-
-
-class TestFolderTrashAfterClose(unittest.TestCase):
-    def test_a_replacement_after_close_deletes_in_place_without_a_thread(self):
-        with tempfile.TemporaryDirectory() as tmp:
-            root = Path(tmp)
-            close_trash(root)
-            folder = root / "output"
-            folder.mkdir()
-            (folder / "late.png").write_bytes(b"png")
-            replace_folder(folder)
-            self.assertEqual(list(folder.iterdir()), [])
-            self.assertFalse((root / TRASH_NAME).exists())
-            names = [t.name for t in threading.enumerate()]
-            self.assertNotIn("pm-reset-trash", names)
-
-
-if __name__ == "__main__":
-    unittest.main()

tests/test_browser_slow_reset_watch.py

@@ -221,12 +221,12 @@ class TestSlowResetOnTheServer(unittest.TestCase):
             self.assertRegex(
                 text, r"kernel: asyncio_\d+ \(0x[0-9a-f]{16}\) state \w, waiting in "
             )
-        # The plain reset seeds a scratch copy (#384), then waits to publish it.
-        self.assertRegex(text, r"line \d+ in _publish\n")
+        # The plain reset's seed() waits on its first DELETE.
+        self.assertRegex(text, r"line \d+ in seed\n")
         self.assertIn("POST /__test__/reset answered 200 after", text)
         self.assertRegex(
             text,
-            r"executor call ScratchSeeder.run: queued \d+\.\d+s, ran \d+\.\d+s",
+            r"executor call create_app.<locals>.seed: queued \d+\.\d+s, ran \d+\.\d+s",
         )
 
 

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,10 +1,8 @@
 """PromptManagerLogger: memory buffer, file rotation, level changes, truncation."""
 
 import logging
-import ntpath
 import os
 import pathlib
-import stat
 import sys
 import tempfile
 import threading
@@ -13,20 +11,14 @@ import unittest.mock as mock
 
 sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
 
-from utils import logging_config  # noqa: E402
-from utils.logging_config import (  # noqa: E402
+from utils import logging_config
+from utils.logging_config import (
     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."""
@@ -40,7 +32,6 @@ 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()
@@ -62,21 +53,12 @@ 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.filters.clear()
-            handler.addFilter(allowed)
+            handler.addFilter(lambda record: record.thread == this_thread)
 
     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()
 
@@ -193,14 +175,15 @@ class TestFilesAndRotation(IsolatedLoggerTestCase):
 
     def test_unreadable_files_are_skipped_in_the_listing(self):
         self.logger.info("something")
-        original_lstat = os.lstat
+        original_stat = pathlib.Path.stat
 
-        def lstat_failing_for_logs(path, *args, **kwargs):
-            if str(path).endswith(".log"):
+        def stat_failing_for_logs(path, *args, **kwargs):
+            if path.suffix == ".log":
                 raise OSError("denied")
-            return original_lstat(path, *args, **kwargs)
+            return original_stat(path, *args, **kwargs)
 
-        with mock.patch.object(logging_config.os, "lstat", lstat_failing_for_logs):
+        with mock.patch.object(pathlib.Path, "stat", autospec=True) as stat:
+            stat.side_effect = stat_failing_for_logs
             self.assertEqual(self.manager.get_log_files(), [])
 
     def test_truncate_does_not_roll_the_log_over_while_deleting_backups(self):
@@ -245,522 +228,10 @@ class TestFilesAndRotation(IsolatedLoggerTestCase):
 
     def test_truncate_reports_files_it_cannot_remove(self):
         self._rotate()
-
-        def busy(path, *args, **kwargs):
-            raise PermissionError(13, "busy", str(path))
-
-        with mock.patch.object(logging_config.os, "unlink", busy):
+        with mock.patch.object(pathlib.Path, "unlink", side_effect=OSError("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):

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,12 +10,9 @@ 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
@@ -23,160 +20,6 @@ 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."""
@@ -244,9 +87,6 @@ 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()
@@ -398,14 +238,6 @@ 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.
 
@@ -413,27 +245,26 @@ 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 (a link's own size: links are not followed)
+            - size: File size in bytes
             - 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 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"],
-                }
-            )
+
+        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}")
 
         # Sort by modification time, newest first
         log_files.sort(key=lambda x: x["modified"], reverse=True)
@@ -471,44 +302,14 @@ class PromptManagerLogger:
             self.logger.error(f"Error reading log file {filename}: {e}")
             raise
 
-    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:
+    def _truncate_main_log(self, log_file: Path) -> 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. 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.
+        keeps position and size in step.
         """
         handled = False
         target = os.path.normcase(os.path.realpath(str(log_file)))
@@ -519,8 +320,6 @@ 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()
@@ -530,65 +329,14 @@ class PromptManagerLogger:
             finally:
                 handler.release()
         if not handled:
-            flags = os.O_WRONLY | os.O_TRUNC | getattr(os, "O_NOFOLLOW", 0)
-            os.close(os.open(str(log_file), flags))
-
-    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}
+            with open(log_file, "w", encoding="utf-8", errors="replace") as f:
+                f.write("")
 
     def truncate_logs(self) -> Dict[str, Any]:
-        """Empty the active log and delete its rotated copies (Rule R).
+        """Truncate all log files.
 
-        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.
+        Clears the main log file and deletes rotated log files.
+        Also clears the memory buffer.
 
         Returns:
             Dictionary with:
@@ -597,31 +345,28 @@ class PromptManagerLogger:
         """
         results = {"truncated": [], "errors": []}
 
-        with self._files_lock:
-            held = self._acquire_file_handlers()
+        # 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:
             try:
-                rows = scan_log_folder(self.log_dir)
-                active = [row for row in rows if row["kind"] == "active"]
-                rotated = sorted(
-                    (row for row in rows if row["kind"] == "rotated"),
-                    key=rotated_sort_key,
+                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)}"
                 )
-                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)
+
+        # Clear memory buffer
+        with self._buffer_lock:
+            self._log_buffer.clear()
 
         if results["truncated"]:
             self.logger.info("Truncated log files: %s", ", ".join(results["truncated"]))
@@ -664,9 +409,7 @@ class PromptManagerLogger:
                 "backup_count",
             ]
         ):
-            # No cleanup may hold the old handler while it is replaced.
-            with self._files_lock:
-                self._setup_loggers()
+            self._setup_loggers()
 
         self.logger.info(f"Updated logging configuration: {new_config}")
 
@@ -697,9 +440,7 @@ class PromptManagerLogger:
                 level = entry["level"]
                 level_counts[level] = level_counts.get(level, 0) + 1
 
-        # 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"]
+        log_files = self.get_log_files()
         total_log_size = sum(f["size"] for f in log_files)
 
         return {

utils/version.py

@@ -1,18 +0,0 @@
-"""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"