Merge remote-tracking branch 'origin/main' into fix/379-per-image-tagging-pr1

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/inspection_service.py

@@ -129,8 +129,9 @@ def inspect_current_file(
         profile = (
             PREVIEW_INSPECTION_PROFILE if _in_host_temp(path) else INSPECTION_PROFILE
         )
+        # The file name anchors on the saver that wrote it (#381).
         evidence = inspect_prompt_evidence(
-            inspection.metadata, role_saved, profile=profile
+            inspection.metadata, role_saved, profile=profile, file_name=path.name
         )
     return {
         "file": {

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)