fix(test): remove /dev/shm reset scratch left by killed browser servers (#384)

A server killed outright never runs its cleanup, and tmpfs keeps the
scratch folder in memory until reboot: the Python suite's server tests
left one per killed server. Scratch folders now carry the owner's pid,
and each new one first removes those whose process is gone (POSIX only).

.github/branch-protection.json

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

.github/workflows/code-quality.yml

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

.github/workflows/test.yml

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

.github/workflows/viewerjs-check.yml

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

__init__.py

@@ -23,24 +23,20 @@ Features:
 """
 
 import atexit
-import re
 import threading
 import time
 from pathlib import Path
 
 
 def get_version():
-    """Parse version from pyproject.toml"""
+    """Parse version from pyproject.toml (``utils/version.py``)."""
+    from .utils.version import project_version
+
     try:
-        pyproject_path = Path(__file__).parent / "pyproject.toml"
-        if pyproject_path.exists():
-            content = pyproject_path.read_text()
-            match = re.search(r'version\s*=\s*["\']([^"\']+)["\']', content)
-            if match:
-                return match.group(1)
+        root = Path(__file__).parent
     except Exception:
-        pass
-    return "unknown"
+        return "unknown"
+    return project_version(root)
 
 
 from .prompt_manager import (  # noqa: E402

py/api/__init__.py

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

py/api/admin.py

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

py/api/diagnostics_report.py

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

py/api/log_cleanup.py

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

py/api/logging_routes.py

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

py/api/maintenance.py

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

tests/browser/fast_reset.py

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

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

@@ -0,0 +1,279 @@
+"""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,8 +19,10 @@ 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 (
@@ -225,6 +227,10 @@ 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]
@@ -238,6 +244,7 @@ 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,
@@ -245,7 +252,7 @@ def create_app(root: Path) -> web.Application:
         extra_prompts: int = 0,
         extra_variants: int = 0,
     ) -> None:
-        with api.db.model.get_connection() as conn:
+        with target.db.model.get_connection() as conn:
             for table in (
                 "saved_views",
                 "image_favorites",
@@ -264,8 +271,7 @@ def create_app(root: Path) -> web.Application:
             ):
                 conn.execute(f"DELETE FROM {table}")
         for name in ("output", "input"):
-            shutil.rmtree(root / name)
-            (root / name).mkdir()
+            replace_folder(root / name)
         for index, (directory, text, tag) in enumerate(
             [
                 ("output", "Cat portrait", "cats"),
@@ -308,28 +314,30 @@ 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 = api.db.save_prompt(
+            pid = target.db.save_prompt(
                 text, tags=[tag], prompt_hash=generate_prompt_hash(text)
             )
-            api.db.link_image_to_prompt(pid, str(image_path))
+            target.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)
-                    api.db.link_image_to_prompt(pid, str(extra))
+                    target.db.link_image_to_prompt(pid, str(extra))
         if native_library:
-            positive = api.db.get_prompt_by_hash(generate_prompt_hash("Cat portrait"))
+            positive = target.db.get_prompt_by_hash(
+                generate_prompt_hash("Cat portrait")
+            )
             positive_id = positive["id"]
-            api.db.update_prompt_metadata(
+            target.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 = api.db.save_prompt_pair(
+                pair_id = target.db.save_prompt_pair(
                     "Cat portrait",
                     negative,
                     title=f"Portrait variant {index + 1}",
@@ -337,28 +345,30 @@ def create_app(root: Path) -> web.Application:
                     tags=["studio"] if index == 0 else ["outdoors"],
                 )
                 variants.append(pair_id)
-            with api.db.model.get_connection() as conn:
+            with target.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]):
-                api.db.link_image_to_pair(
+                target.db.link_image_to_pair(
                     variants[index], image["id"], {"source": "fixture", "anchor": index}
                 )
-            api.db.link_image_to_pair(
+            target.db.link_image_to_pair(
                 variants[0], images[0]["id"], {"source": "fixture second evidence"}
             )
-            api.db.record_prompt_pair_use(variants[0], times=3)
+            target.db.record_prompt_pair_use(variants[0], times=3)
             for number in range(extra_variants):
-                api.db.save_prompt_pair("Cat portrait", f"Negative extra {number:04}")
+                target.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)
-            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")
+            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")
             for number in range(extra_prompts):
-                api.db.save_prompt(
+                target.db.save_prompt(
                     f"Library extra {number:04}", category="Bulk", tags=["bulk"]
                 )
         if prompt_inspector:
@@ -379,8 +389,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).
-        api._generate_gallery_thumbnails_sync(
-            api._get_all_output_dirs(), ThumbnailSettings("webp", 80, 150)
+        target._generate_gallery_thumbnails_sync(
+            target._get_all_output_dirs(), ThumbnailSettings("webp", 80, 150)
         )
         if include_video:
             # Seed after thumbnail generation to exercise a real unthumbnailed video.
@@ -388,8 +398,8 @@ def create_app(root: Path) -> web.Application:
             shutil.copyfile(
                 Path(__file__).resolve().parents[1] / "fixtures" / "tiny.webm", video
             )
-            api.db.link_image_to_prompt(video_prompt_id, str(video))
-        api.invalidate_gallery_cache()
+            target.db.link_image_to_prompt(video_prompt_id, str(video))
+        target.invalidate_gallery_cache()
 
     history_fixture = None
 
@@ -434,7 +444,10 @@ def create_app(root: Path) -> web.Application:
             raise web.HTTPBadRequest()
         if chronological_history:
             manifest = await api._run_in_executor(
-                lambda: seed_workspace_history(api, root, history_runs=history_runs)
+                seeder.run,
+                lambda target: seed_workspace_history(
+                    target, root, history_runs=history_runs
+                ),
             )
             history_fixture = manifest
             return web.json_response({"success": True, **manifest})
@@ -492,25 +505,31 @@ def create_app(root: Path) -> web.Application:
             raise web.HTTPBadRequest()
         if workspace_visual:
             history_fixture = None
-            await api._run_in_executor(seed_workspace_visual, api, root)
+            await api._run_in_executor(
+                seeder.run, lambda target: seed_workspace_visual(target, root)
+            )
             return web.json_response({"success": True})
         if full_inspector:
             history_fixture = None
             await api._run_in_executor(
-                lambda: seed_workspace_inspector(
-                    api, root, extra_images=extra, include_video=include_video
-                )
+                seeder.run,
+                lambda target: seed_workspace_inspector(
+                    target, root, extra_images=extra, include_video=include_video
+                ),
             )
             return web.json_response({"success": True})
         history_fixture = None
         await api._run_in_executor(
-            seed,
-            extra,
-            include_video,
-            prompt_inspector,
-            native_library,
-            extra_prompts,
-            extra_variants,
+            seeder.run,
+            lambda target: seed(
+                target,
+                extra,
+                include_video,
+                prompt_inspector,
+                native_library,
+                extra_prompts,
+                extra_variants,
+            ),
         )
         return web.json_response({"success": True})
 
@@ -550,14 +569,15 @@ def create_app(root: Path) -> web.Application:
         reveal_stub.reset()
         history_fixture = None
         manifest = await api._run_in_executor(
-            lambda: seed_workspace_concept(
-                api,
+            seeder.run,
+            lambda target: seed_workspace_concept(
+                target,
                 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})
 
@@ -625,9 +645,12 @@ 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()
+    seed(api)
     # 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_concept_fixture.py

@@ -16,7 +16,6 @@ import json
 import math
 import os
 from pathlib import Path
-import shutil
 import time
 from types import SimpleNamespace
 from typing import TYPE_CHECKING
@@ -34,6 +33,7 @@ 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,8 +300,7 @@ def _clear(api: PromptManagerAPI, root: Path) -> None:
         for table in _TABLES:
             conn.execute(f"DELETE FROM {table}")
     for name in ("output", "input"):
-        shutil.rmtree(root / name)
-        (root / name).mkdir()
+        replace_folder(root / name)
 
 
 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,8 +51,7 @@ def seed_library_editor(api: PromptManagerAPI, root: Path) -> dict:
         for table in _TABLES:
             conn.execute(f"DELETE FROM {table}")
     for name in ("output", "input"):
-        shutil.rmtree(root / name)
-        (root / name).mkdir()
+        replace_folder(root / name)
     epoch = db.get_execution_epoch()
     receipt = db.create_library_recipe(
         EDITOR_POSITIVE,

tests/browser/workspace_review_fixture.py

@@ -52,6 +52,7 @@ 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
@@ -225,8 +226,7 @@ def seed_review(
             conn.execute(f"DELETE FROM {table}")
     api.close_annotation_captures()
     for name in ("output", "input"):
-        shutil.rmtree(root / name)
-        (root / name).mkdir()
+        replace_folder(root / name)
     shutil.rmtree(root / "temp" / "review", ignore_errors=True)
     output = root / "output"
     files = []

tests/browser/workspace_tagging_fixture.py

@@ -19,6 +19,7 @@ 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
@@ -98,8 +99,7 @@ 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"):
-        shutil.rmtree(root / name)
-        (root / name).mkdir()
+        replace_folder(root / name)
     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,8 +129,7 @@ def seed_workspace_visual(api: PromptManagerAPI, root: Path) -> None:
         ):
             conn.execute(f"DELETE FROM {table}")
     for name in ("output", "input"):
-        shutil.rmtree(root / name)
-        (root / name).mkdir()
+        replace_folder(root / name)
     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,8 +120,7 @@ def seed_vocabulary(
         for table in _TABLES:
             conn.execute(f"DELETE FROM {table}")
     for name in ("output", "input"):
-        shutil.rmtree(root / name)
-        (root / name).mkdir()
+        replace_folder(root / name)
     epoch = db.get_execution_epoch()
     recipes = []
     for recipe in CONCEPT_RECIPES:

tests/test_api_diagnostics_report.py

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

tests/test_api_log_cleanup.py

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

tests/test_api_logging.py

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

tests/test_browser_fast_reset.py

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

@@ -0,0 +1,67 @@
+"""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's seed() waits on its first DELETE.
-        self.assertRegex(text, r"line \d+ in seed\n")
+        # The plain reset seeds a scratch copy (#384), then waits to publish it.
+        self.assertRegex(text, r"line \d+ in _publish\n")
         self.assertIn("POST /__test__/reset answered 200 after", text)
         self.assertRegex(
             text,
-            r"executor call create_app.<locals>.seed: queued \d+\.\d+s, ran \d+\.\d+s",
+            r"executor call ScratchSeeder.run: queued \d+\.\d+s, ran \d+\.\d+s",
         )
 
 

tests/test_execution_history_hooks.py

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

tests/test_execution_output_hooks.py

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

tests/test_execution_output_writer.py

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

tests/test_utils_logging_config.py

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

tests/vendor/browser-image.test.mjs

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

utils/logging_config.py

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

utils/version.py

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