feat(system): read-only health check and truthful overview counts

POST /prompt_manager/system/health runs the health check off-loop as a
read registry job of kind "health" (one at a time, beside a scan, not
cancellable, no file outcomes) and keeps the last report; GET returns it
or null. The check reads one snapshot on its own mode=ro connection,
closed in finally: integrity_check(100), foreign_key_check, stored hashes
against utils.hashing, duplicate pair rows, the legacy consistency
queries (now shared with check_consistency), and image links whose file
is gone, counted only under readable sources. An unavailable source, or
the absent Needs-review queue, makes its check unavailable and the
report incomplete instead of producing a count.

GET /prompt_manager/system/overview returns Library-eligible positive
prompts and active negative variants, indexed images, the last health
summary or null, backup null, the Needs-review count null, the effective
sources with whether they were auto-detected, and the thumbnail plan's
missing count. GET /prompt_manager/system/statistics lists runs and
average rating per category of Library parents, with the /stats totals.

database/health.py

@@ -0,0 +1,265 @@
+"""The read-only health check behind System › Overview and Maintenance.
+
+``run_health_checks`` opens its own ``mode=ro`` connection, reads one
+snapshot and closes the connection in ``finally``; it never writes. Each
+check reports ``passed``, ``issues`` (with an exact ``count``),
+``unavailable`` (it could not be checked here, so ``count`` is None) or
+``failed`` (it raised). At most ``MAX_DETAIL`` detail strings are returned per
+check, and none of them carries a server path.
+"""
+
+import datetime
+import logging
+import os
+import sqlite3
+from pathlib import Path
+
+try:
+    from ..utils.hashing import generate_prompt_hash
+    from .models import normalize_image_path
+except ImportError:
+    from utils.hashing import generate_prompt_hash
+    from database.models import normalize_image_path
+
+logger = logging.getLogger("prompt_manager.database.health")
+
+MAX_DETAIL = 10  # detail strings returned per check
+INTEGRITY_LIMIT = 100  # PRAGMA integrity_check(N): problems SQLite reports
+BATCH_ROWS = 100  # image rows read per fetch
+REVIEW_UNAVAILABLE = "Needs review is not available yet."
+CHECK_FAILED = "The check could not finish."
+# Check ids and the labels of the concept's health report, in its order;
+# pair uniqueness and consistency are the added rows (D63, now D95).
+CHECKS = (
+    ("integrity", "Integrity check"),
+    ("foreign_keys", "Foreign keys & pair references"),
+    ("canonical_hashes", "Canonical hashes"),
+    ("pair_uniqueness", "Pair uniqueness"),
+    ("missing_files", "Missing files"),
+    ("ambiguous_metadata", "Ambiguous metadata"),
+    ("consistency", "Consistency"),
+)
+_DATABASE_CHECKS = (
+    "integrity",
+    "foreign_keys",
+    "canonical_hashes",
+    "pair_uniqueness",
+    "missing_files",
+    "consistency",
+)
+
+
+def iter_consistency_issues(conn):
+    """The legacy consistency issues: tag and image rows of a missing prompt."""
+    for ref in conn.execute(
+        "SELECT pt.prompt_id FROM prompt_tags pt "
+        "LEFT JOIN prompts p ON pt.prompt_id = p.id WHERE p.id IS NULL"
+    ):
+        yield f"prompt_tags entry references non-existent prompt {ref[0]}"
+    for ref in conn.execute(
+        "SELECT gi.id, gi.prompt_id FROM generated_images gi "
+        "LEFT JOIN prompts p ON gi.prompt_id = p.id WHERE p.id IS NULL"
+    ):
+        yield f"Image {ref[0]} references non-existent prompt {ref[1]}"
+
+
+def _result(count, detail, *, status=None):
+    return {
+        "status": status or ("issues" if count else "passed"),
+        "count": count,
+        "detail": detail[:MAX_DETAIL],
+    }
+
+
+def _counted(lines):
+    """(count, first MAX_DETAIL lines) of an iterable, consumed once."""
+    count, detail = 0, []
+    for line in lines:
+        count += 1
+        if len(detail) < MAX_DETAIL:
+            detail.append(line)
+    return _result(count, detail)
+
+
+def _integrity(conn, _roots):
+    rows = [
+        row[0] for row in conn.execute(f"PRAGMA integrity_check({INTEGRITY_LIMIT})")
+    ]
+    if rows == ["ok"]:
+        return _result(0, [])
+    return _result(len(rows), rows)
+
+
+def _foreign_keys(conn, _roots):
+    return _counted(
+        f"{table} row {rowid} references a missing {parent} row"
+        for table, rowid, parent, _fkid in conn.execute("PRAGMA foreign_key_check")
+    )
+
+
+def _prompt_hash(text):
+    return generate_prompt_hash(text) if isinstance(text, str) else None
+
+
+def _canonical_hashes(conn, _roots):
+    conn.create_function("pm_prompt_hash", 1, _prompt_hash, deterministic=True)
+    return _counted(
+        f"Prompt {row[0]}: the stored hash does not match its text"
+        for row in conn.execute(
+            "SELECT id FROM prompts WHERE hash IS NOT pm_prompt_hash(text) ORDER BY id"
+        )
+    )
+
+
+def _pair_uniqueness(conn, _roots):
+    groups = conn.execute(
+        "SELECT positive_prompt_id,negative_prompt_id,COUNT(*) FROM prompt_pairs "
+        "GROUP BY positive_prompt_id,negative_prompt_id HAVING COUNT(*)>1 "
+        "ORDER BY positive_prompt_id,negative_prompt_id"
+    ).fetchall()
+    detail = [
+        f"Positive prompt {positive} with "
+        + ("no negative" if negative is None else f"negative {negative}")
+        + f": {rows} pair rows"
+        for positive, negative, rows in groups
+    ]
+    # The surplus rows: what keeping one row per pair would leave out.
+    return _result(sum(rows - 1 for _p, _n, rows in groups), detail)
+
+
+def _consistency(conn, _roots):
+    return _counted(iter_consistency_issues(conn))
+
+
+def _readable(root):
+    return os.path.isdir(root) and os.access(root, os.R_OK | os.X_OK)
+
+
+def _root_prefixes(roots):
+    return [normalize_image_path(root).rstrip("\\/") + os.sep for root in roots]
+
+
+def _under(path, prefixes):
+    return any(path.startswith(prefix) for prefix in prefixes)
+
+
+def _missing_files(conn, roots):
+    """Links whose file is gone, counted only under readable sources.
+
+    A source that is not readable (an unmounted drive) makes the whole check
+    unavailable rather than turning its links into a count; links outside
+    every source are named in the detail, not counted.
+    """
+    if not roots:
+        return _result(None, ["No image source is available."], status="unavailable")
+    unavailable = [root for root in roots if not _readable(root)]
+    if unavailable:
+        noun = "source is" if len(unavailable) == 1 else "sources are"
+        return _result(
+            None,
+            [
+                f"{len(unavailable)} image {noun} unavailable, "
+                "so missing files were not counted."
+            ],
+            status="unavailable",
+        )
+    prefixes = _root_prefixes(roots)
+    count, outside, detail = 0, 0, []
+    cursor = conn.execute(
+        "SELECT id,image_path,filename,file_path FROM generated_images ORDER BY id"
+    )
+    while rows := cursor.fetchmany(BATCH_ROWS):
+        for image_id, image_path, filename, file_path in rows:
+            spelled = file_path or normalize_image_path(image_path)
+            if not _under(spelled, prefixes) and not _under(
+                normalize_image_path(os.path.realpath(image_path)), prefixes
+            ):
+                outside += 1
+            elif not os.path.exists(image_path):
+                count += 1
+                if len(detail) < MAX_DETAIL:
+                    name = filename or os.path.basename(image_path)
+                    detail.append(f"Image {image_id}: {name} is missing")
+    result = _result(count, detail)
+    if outside and len(result["detail"]) < MAX_DETAIL:
+        links = "link" if outside == 1 else "links"
+        verb = "was" if outside == 1 else "were"
+        result["detail"].append(
+            f"{outside} {links} outside the image sources {verb} not checked."
+        )
+    return result
+
+
+def _ambiguous(review_count):
+    if review_count is None:
+        return _result(None, [REVIEW_UNAVAILABLE], status="unavailable")
+    return _result(review_count, [])
+
+
+def overall_status(checks) -> str:
+    """``passed`` only when every check passed; any check not run is incomplete."""
+    statuses = {check["status"] for check in checks}
+    if statuses & {"unavailable", "failed"}:
+        return "incomplete"
+    return "issues" if "issues" in statuses else "passed"
+
+
+def _failed():
+    return _result(None, [CHECK_FAILED], status="failed")
+
+
+def _run_database_checks(db_path, roots):
+    """Each database check's result (None when it raised), from one snapshot."""
+    results = dict.fromkeys(_DATABASE_CHECKS)
+    try:
+        conn = sqlite3.connect(Path(db_path).as_uri() + "?mode=ro", uri=True)
+    except sqlite3.Error:
+        logger.exception("Health check could not open the database")
+        return results
+    try:
+        conn.execute("BEGIN")  # one read snapshot for every check
+        for check_id in _DATABASE_CHECKS:
+            runner = globals()["_" + check_id]  # looked up per run (tests patch it)
+            try:
+                results[check_id] = runner(conn, roots)
+            except (sqlite3.Error, OSError):
+                logger.exception(f"Health check {check_id} failed")
+    except sqlite3.Error:
+        logger.exception("Health check could not read the database")
+    finally:
+        conn.close()
+    return results
+
+
+class HealthOperations:
+    """Mixin for PromptDatabase: the read-only System health check."""
+
+    def run_health_checks(self, roots, *, review_count=None) -> dict:
+        """Run every check and return a ``HealthReport``.
+
+        Args:
+            roots: The image source directories to judge missing files under:
+                the effective sources plus any configured source that is not
+                available (which makes ``missing_files`` unavailable).
+            review_count: The Needs-review pending count, or None while no
+                review queue exists (``ambiguous_metadata`` is then unavailable).
+        """
+        if review_count is not None and (
+            type(review_count) is not int or review_count < 0
+        ):
+            raise ValueError("review_count must be a non-negative integer or None")
+        roots = [str(root) for root in roots]
+        database = _run_database_checks(self.model.db_path, roots)
+        checks = []
+        for check_id, label in CHECKS:
+            if check_id == "ambiguous_metadata":
+                result = _ambiguous(review_count)
+            else:
+                result = database[check_id] or _failed()
+            checks.append({"id": check_id, "label": label, **result})
+        return {
+            "success": True,
+            "checked_at": datetime.datetime.now(datetime.timezone.utc).isoformat(),
+            "status": overall_status(checks),
+            "checks": checks,
+        }

database/library.py

@@ -628,3 +628,49 @@ class LibraryOperations:
                 "kind": "prompt-manager-library",
                 "positives": [owners[key] for key in order],
             }
+
+    def library_counts(self) -> dict:
+        """System › Overview counts from one snapshot, writing nothing.
+
+        ``positive_prompts`` and ``negative_variants`` are the unfiltered
+        Library's ``total`` and ``variant_total``: eligible parents and active
+        pairs (every pair's positive is eligible). ``images`` counts indexed
+        ``generated_images`` rows, not files on disk.
+        """
+        with self.library_snapshot() as (conn, _epoch, _ceiling):
+            row = conn.execute(
+                "SELECT (SELECT COUNT(*) FROM prompts p WHERE " + _ELIGIBLE + "),"
+                "(SELECT COUNT(*) FROM prompt_pairs WHERE archived=0),"
+                "(SELECT COUNT(*) FROM generated_images)"
+            ).fetchone()
+        return {
+            "positive_prompts": row[0],
+            "negative_variants": row[1],
+            "images": row[2],
+        }
+
+    def usage_by_category(self) -> list[dict]:
+        """Runs and average rating of the Library's parents, per category.
+
+        Only eligible parents count, so a negative-only prompt's runs (which
+        repeat its pairs' runs) are never added twice. A blank category groups
+        with no category as ``None``. Unrated parents are left out of the
+        average, which is ``None`` when no parent is rated. Most runs first.
+        """
+        with self.library_snapshot() as (conn, _epoch, _ceiling):
+            rows = conn.execute(
+                "SELECT * FROM (SELECT NULLIF(TRIM(p.category),'') AS category,"
+                "COALESCE(SUM(p.run_count),0) AS runs,AVG(p.rating) AS rating "
+                "FROM prompts p WHERE " + _ELIGIBLE + " GROUP BY 1) "
+                "ORDER BY runs DESC,category IS NULL,category"
+            ).fetchall()
+        return [
+            {
+                "category": row["category"],
+                "runs": row["runs"],
+                "average_rating": (
+                    None if row["rating"] is None else round(row["rating"], 2)
+                ),
+            }
+            for row in rows
+        ]

database/operations.py

@@ -20,6 +20,7 @@ from .execution_history import ExecutionHistoryOperations
 from .execution_contexts import ExecutionContextOperations, LIVE_HINT_CTES
 from .execution_outputs import ExecutionOutputOperations
 from .history_reads import HistoryReadOperations
+from .health import HealthOperations, iter_consistency_issues
 
 # Import logging system
 try:
@@ -342,6 +343,7 @@ class PromptDatabase(
     ImageFavoriteOperations,
     AnnotationOperations,
     SavedViewOperations,
+    HealthOperations,
 ):
     """Database operations class for managing prompts."""
 
@@ -2570,33 +2572,13 @@ class PromptDatabase(
             return cursor.rowcount
 
     def check_consistency(self) -> List[str]:
-        """Run consistency checks and return a list of issue descriptions."""
-        issues: List[str] = []
-        with self.model.get_connection() as conn:
-            # Check for orphaned prompt_tags entries
-            cursor = conn.execute("""
-                SELECT pt.prompt_id, pt.tag_id FROM prompt_tags pt
-                LEFT JOIN prompts p ON pt.prompt_id = p.id
-                WHERE p.id IS NULL
-            """)
-            for ref in cursor.fetchall():
-                issues.append(
-                    "prompt_tags entry references non-existent prompt "
-                    f"{ref['prompt_id']}"
-                )
+        """Run consistency checks and return a list of issue descriptions.
 
-            # Check for orphaned image entries
-            cursor = conn.execute("""
-                SELECT gi.id, gi.prompt_id FROM generated_images gi
-                LEFT JOIN prompts p ON gi.prompt_id = p.id
-                WHERE p.id IS NULL
-            """)
-            for ref in cursor.fetchall():
-                issues.append(
-                    f"Image {ref['id']} references non-existent prompt "
-                    f"{ref['prompt_id']}"
-                )
-        return issues
+        The same queries back the System health check's Consistency row
+        (database/health.py).
+        """
+        with self.model.get_connection() as conn:
+            return list(iter_consistency_issues(conn))
 
     def _image_row_to_dict(self, row: sqlite3.Row) -> Dict[str, Any]:
         """

py/api/__init__.py

@@ -8,6 +8,7 @@ Split into domain-specific mixins for maintainability:
 - 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
+- SystemRoutesMixin: System health check, Overview and usage statistics
 """
 
 import asyncio
@@ -35,6 +36,7 @@ from .admin import AdminRoutesMixin
 from .settings_routes import SettingsRoutesMixin
 from .job_routes import JobRoutesMixin
 from .image_tag_jobs import ImageTagJobsMixin
+from .system import SystemRoutesMixin
 from .jobs import JobRegistry
 from .logging_routes import LoggingRoutesMixin
 from .autotag_routes import AutotagRoutesMixin
@@ -265,6 +267,7 @@ class PromptManagerAPI(
     SettingsRoutesMixin,
     JobRoutesMixin,
     ImageTagJobsMixin,
+    SystemRoutesMixin,
     LoggingRoutesMixin,
     AutotagRoutesMixin,
     LoraIntegrationMixin,
@@ -321,6 +324,8 @@ class PromptManagerAPI(
         self._scan_job = None
         # The current duplicate scan (a py/api/jobs.Job outside the registry).
         self._duplicate_job = None
+        # The last System health report this session (py/api/system.py).
+        self._health_report = None
 
         # Run cleanup on initialization to remove any existing duplicates
         try:
@@ -478,6 +483,7 @@ class PromptManagerAPI(
         self._register_admin_routes(routes)
         self._register_settings_routes(routes)
         self._register_job_routes(routes)
+        self._register_system_routes(routes)
         self._register_logging_routes(routes)
         self._register_autotag_routes(routes)
         self._register_lora_routes(routes)
@@ -719,6 +725,13 @@ class PromptManagerAPI(
         Returns:
             List[Path]: Validated directory paths, possibly empty.
         """
+        return self._output_dirs_with_origin()[0]
+
+    def _output_dirs_with_origin(self):
+        """``(dirs, auto_detected)``: the ``_get_all_output_dirs`` list, and
+        whether it comes from detection or ComfyUI's output rather than from
+        a valid configured root (False when there is no directory at all).
+        """
         from ..config import GalleryConfig
 
         configured = list(GalleryConfig.MONITORING_DIRECTORIES)
@@ -737,15 +750,16 @@ class PromptManagerAPI(
             output_dirs.append(Path(real))
 
         if output_dirs:
-            return output_dirs
+            return output_dirs, False
 
         if configured:
             # Every configured root was rejected: never let auto-detection
             # resurrect it, serve ComfyUI's own output directory instead.
-            return self._comfyui_output_dir_list()
-
-        fallback = self._find_comfyui_output_dir()
-        return [Path(fallback)] if fallback else []
+            fallback_dirs = self._comfyui_output_dir_list()
+        else:
+            fallback = self._find_comfyui_output_dir()
+            fallback_dirs = [Path(fallback)] if fallback else []
+        return fallback_dirs, bool(fallback_dirs)
 
     def _comfyui_output_dir_list(self):
         """ComfyUI's output directory as a one-element list, or [] if unknown."""

py/api/jobs.py

@@ -88,6 +88,9 @@ DUPLICATE_CLEANUP_JOB_KIND = "duplicate_cleanup"
 THUMBNAILS_JOB_KIND = "thumbnails"
 LORA_IMPORT_JOB_KIND = "lora_import"
 AUTOTAG_BATCH_JOB_KIND = "autotag"
+# The System health check (py/api/system.py): a read-only ``read`` job with
+# no file outcomes, so it may run beside a scan.
+HEALTH_JOB_KIND = "health"
 
 
 def resolve_admission(admission=None, record_changing=None) -> str:

py/api/system.py

@@ -0,0 +1,177 @@
+"""System routes: the read-only health check, the Overview and usage statistics.
+
+- ``POST /prompt_manager/system/health`` starts the health check
+  (``database/health.py``) off-loop as a ``read`` job of kind ``health``: one
+  at a time, beside a scan, not cancellable and with no file outcomes. The
+  API keeps the last report for this session; ``GET`` returns it or null.
+- ``GET /prompt_manager/system/overview`` is the System › Overview aggregate.
+  Every value is real or null: there is no backup status until backups report
+  one, no health result until a check ran, and no Needs-review count until
+  the review queue exists.
+- ``GET /prompt_manager/system/statistics`` lists runs and ratings per
+  category, with the legacy ``/stats`` totals.
+
+New routes answer errors with ``{success:false, code, error}``.
+"""
+
+import asyncio
+import os
+
+from aiohttp import web
+
+from .job_routes import busy_response
+from .jobs import HEALTH_JOB_KIND, JobBusy
+
+HEALTH_JOB_TITLE = "Health check"
+_HEALTH_STATUS_LINES = {
+    "passed": "Health check passed",
+    "issues": "Health check found issues",
+    "incomplete": "Health check incomplete",
+}
+
+
+def _failure(message, status=500, code=None):
+    return web.json_response(
+        {"success": False, "code": code, "error": message}, status=status
+    )
+
+
+def health_summary(report):
+    """What the Overview shows of a report: when, the outcome, checks with issues."""
+    if report is None:
+        return None
+    return {
+        "checked_at": report["checked_at"],
+        "status": report["status"],
+        "issues": sum(1 for check in report["checks"] if check["status"] == "issues"),
+    }
+
+
+class SystemRoutesMixin:
+    """Mixin providing the /prompt_manager/system endpoints."""
+
+    def _register_system_routes(self, routes):
+        @routes.post("/prompt_manager/system/health")
+        async def start_health_check_route(request):
+            return await self.start_health_check(request)
+
+        @routes.get("/prompt_manager/system/health")
+        async def health_report_route(request):
+            return await self.get_health_report(request)
+
+        @routes.get("/prompt_manager/system/overview")
+        async def system_overview_route(request):
+            return await self.get_system_overview(request)
+
+        @routes.get("/prompt_manager/system/statistics")
+        async def system_statistics_route(request):
+            return await self.get_system_statistics(request)
+
+    # ── health ────────────────────────────────────────────────────────
+
+    def _health_roots(self):
+        """The sources missing files are judged under.
+
+        The effective sources, plus every configured source that is not an
+        existing directory (an unmounted drive): the check then reports
+        missing files as unavailable instead of counting that source's links.
+        """
+        from ..config import GalleryConfig
+
+        roots = [str(path) for path in self._get_all_output_dirs()]
+        for configured in GalleryConfig.MONITORING_DIRECTORIES:
+            try:
+                resolved = GalleryConfig.resolve_gallery_root(configured)
+            except (OSError, TypeError, ValueError):
+                continue
+            if not os.path.isdir(resolved):
+                roots.append(resolved)
+        return roots
+
+    async def start_health_check(self, request):
+        """Start the health check job (202), or 409 naming the running check."""
+        roots = self._health_roots()
+
+        async def run(job):
+            job.update(status_line="Checking the database and image links")
+            report = await self._run_in_executor(self.db.run_health_checks, roots)
+            self._health_report = report
+            job.update(progress=100, status_line=_HEALTH_STATUS_LINES[report["status"]])
+
+        try:
+            job = self._jobs.start(
+                HEALTH_JOB_KIND,
+                HEALTH_JOB_TITLE,
+                admission="read",
+                cancellable=False,
+                scope={},
+                runner=run,
+            )
+        except JobBusy as busy:
+            return busy_response(busy)
+        return web.json_response({"success": True, "job": job.summary()}, status=202)
+
+    async def get_health_report(self, request):
+        return web.json_response({"success": True, "report": self._health_report})
+
+    # ── overview ──────────────────────────────────────────────────────
+
+    def _missing_thumbnails(self, output_dirs):
+        """Pending thumbnails as ``/images/thumbnails/status`` counts them."""
+        if not output_dirs:
+            return None
+        try:
+            plan = self._plan_gallery_thumbnails(output_dirs)
+        except OSError:
+            self.logger.warning("Overview could not plan thumbnails", exc_info=True)
+            return None
+        return len(plan["targets"]) + plan["remaining"]
+
+    async def get_system_overview(self, request):
+        from ..config import GalleryConfig
+
+        try:
+            output_dirs, auto_detected = self._output_dirs_with_origin()
+            counts, missing = await asyncio.gather(
+                self._run_in_executor(self.db.library_counts),
+                self._run_in_executor(self._missing_thumbnails, output_dirs),
+            )
+        except Exception:
+            self.logger.exception("System overview failed")
+            return _failure("Failed to read the overview")
+        return web.json_response(
+            {
+                "success": True,
+                "library": {
+                    "positive_prompts": counts["positive_prompts"],
+                    "negative_variants": counts["negative_variants"],
+                },
+                "images": {"indexed": counts["images"]},
+                "health": health_summary(self._health_report),
+                "backup": None,  # no verified backup status exists yet (S8)
+                "needs_review": {"pending": None},  # no review queue yet (Q6)
+                "sources": {
+                    "configured": len(GalleryConfig.MONITORING_DIRECTORIES),
+                    "effective": len(output_dirs),
+                    "auto_detected": auto_detected,
+                },
+                "thumbnails": {"missing": missing},
+            },
+            headers={"Cache-Control": "no-store"},
+        )
+
+    # ── statistics ────────────────────────────────────────────────────
+
+    async def get_system_statistics(self, request):
+        try:
+            by_category, totals = await asyncio.gather(
+                self._run_in_executor(self.db.usage_by_category),
+                self._run_in_executor(self.db.get_statistics),
+            )
+        except Exception:
+            self.logger.exception("System statistics failed")
+            return _failure("Failed to read statistics")
+        return web.json_response(
+            {"success": True, "by_category": by_category, "totals": totals},
+            headers={"Cache-Control": "no-store"},
+        )

tests/test_database_health.py

@@ -0,0 +1,392 @@
+"""The read-only health check (System › Overview and Maintenance).
+
+``run_health_checks`` reads one snapshot on its own ``mode=ro`` connection:
+SQLite's integrity and foreign-key checks, stored prompt hashes against
+``utils.hashing``, duplicate pair rows, the legacy consistency queries, image
+links whose file is gone (only under readable sources) and the Needs-review
+count when one is supplied. It never writes, and it never turns a check it
+could not run into a count.
+"""
+
+import hashlib
+import os
+import sqlite3
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+from database import health  # noqa: E402
+from database.operations import PromptDatabase  # noqa: E402
+from utils.hashing import generate_prompt_hash  # noqa: E402
+
+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
+
+CHECK_IDS = [
+    "integrity",
+    "foreign_keys",
+    "canonical_hashes",
+    "pair_uniqueness",
+    "missing_files",
+    "ambiguous_metadata",
+    "consistency",
+]
+
+
+def _digest(path):
+    if not os.path.exists(path):
+        return None
+    with open(path, "rb") as handle:
+        return hashlib.sha256(handle.read()).hexdigest()
+
+
+class HealthTestCase(unittest.TestCase):
+    def setUp(self):
+        self.temp = tempfile.TemporaryDirectory()
+        self.root = Path(os.path.realpath(self.temp.name))
+        self.output = self.root / "output"
+        self.output.mkdir()
+        self.db_path = str(self.root / "prompts.db")
+        self.db = PromptDatabase(self.db_path)
+
+    def tearDown(self):
+        self.db.close_all()
+        assert_closed(self, self.root)
+        self.temp.cleanup()
+
+    def save(self, text, **metadata):
+        return self.db.save_prompt(
+            text, prompt_hash=generate_prompt_hash(text), **metadata
+        )
+
+    def image(self, prompt_id, name, *, exists=True, folder=None):
+        folder = Path(folder or self.output)
+        path = folder / name
+        if exists:
+            path.write_bytes(b"png")
+        return self.db.link_image_to_prompt(prompt_id, str(path))
+
+    def run_checks(self, roots=None, **kwargs):
+        roots = [str(self.output)] if roots is None else roots
+        return self.db.run_health_checks(roots, **kwargs)
+
+    @staticmethod
+    def check(report, check_id):
+        return next(c for c in report["checks"] if c["id"] == check_id)
+
+    def raw(self):
+        """A plain connection for planting damage the app itself never writes."""
+        conn = sqlite3.connect(self.db_path)
+        conn.execute("PRAGMA foreign_keys=OFF")
+        return conn
+
+
+class TestHealthReport(HealthTestCase):
+    def test_clean_library_passes_except_unavailable_review(self):
+        prompt = self.save("A clean prompt")
+        self.db.save_prompt_pair("Cat", "Blur")
+        self.image(prompt, "a.png")
+        report = self.run_checks(review_count=0)
+        self.assertTrue(report["success"])
+        self.assertEqual([c["id"] for c in report["checks"]], CHECK_IDS)
+        self.assertEqual(
+            [c["label"] for c in report["checks"]],
+            [
+                "Integrity check",
+                "Foreign keys & pair references",
+                "Canonical hashes",
+                "Pair uniqueness",
+                "Missing files",
+                "Ambiguous metadata",
+                "Consistency",
+            ],
+        )
+        for check in report["checks"]:
+            self.assertEqual(check["status"], "passed", check)
+            self.assertEqual(check["count"], 0, check)
+        self.assertEqual(report["status"], "passed")
+        self.assertRegex(report["checked_at"], r"^\d{4}-\d\d-\d\dT.*\+00:00$")
+
+    def test_health_check_never_writes(self):
+        prompt = self.save("Untouched")
+        self.image(prompt, "kept.png")
+        self.image(prompt, "gone.png", exists=False)
+        with self.raw() as conn:
+            conn.execute("UPDATE prompts SET hash='stale' WHERE id=?", (prompt,))
+        conn.close()
+        # The app's own connection stays open, as in the running server.
+        self.db.get_statistics()
+        files = [self.db_path, self.db_path + "-wal"]
+        before = [_digest(path) for path in files]
+        shm_size = os.path.getsize(self.db_path + "-shm")
+        media = sorted(os.listdir(self.output))
+        opened = []
+        real_connect = sqlite3.connect
+
+        class Recording(sqlite3.Connection):
+            changes_at_close = None
+
+            def close(self):
+                Recording.changes_at_close = self.total_changes
+                super().close()
+
+        def connect(*args, **kwargs):
+            conn = real_connect(*args, factory=Recording, **kwargs)
+            opened.append((args, kwargs, conn))
+            return conn
+
+        with patch.object(health.sqlite3, "connect", side_effect=connect):
+            report = self.run_checks()
+        self.assertEqual(report["status"], "incomplete")
+        self.assertEqual(len(opened), 1)
+        args, kwargs, conn = opened[0]
+        self.assertTrue(str(args[0]).endswith("?mode=ro"), args)
+        self.assertTrue(kwargs.get("uri"))
+        self.assertEqual(Recording.changes_at_close, 0)
+        with self.assertRaises(sqlite3.ProgrammingError):  # closed in finally
+            conn.execute("SELECT 1")
+        self.assertEqual([_digest(path) for path in files], before)
+        # The -shm holds WAL read marks every reader updates; it never grows.
+        self.assertEqual(os.path.getsize(self.db_path + "-shm"), shm_size)
+        self.assertEqual(sorted(os.listdir(self.output)), media)
+        with self.db.model.get_connection() as live:
+            stored = live.execute(
+                "SELECT hash FROM prompts WHERE id=?", (prompt,)
+            ).fetchone()[0]
+        self.assertEqual(stored, "stale")
+
+    def test_connection_closed_when_a_check_raises(self):
+        self.save("Any")
+        with (
+            patch.object(
+                health,
+                "_canonical_hashes",
+                side_effect=sqlite3.OperationalError("boom"),
+            ),
+            self.assertLogs("prompt_manager.database.health", "ERROR"),
+        ):
+            report = self.run_checks(review_count=0)
+        canonical = self.check(report, "canonical_hashes")
+        self.assertEqual(canonical["status"], "failed")
+        self.assertIsNone(canonical["count"])
+        self.assertEqual(canonical["detail"], ["The check could not finish."])
+        self.assertEqual(self.check(report, "integrity")["status"], "passed")
+        self.assertEqual(report["status"], "incomplete")
+        self.db.close_all()
+        assert_closed(self, self.root)
+
+    def test_unopenable_database_fails_every_database_check(self):
+        missing = PromptDatabase.__new__(PromptDatabase)
+        missing.model = type("M", (), {"db_path": str(self.root / "nope.db")})()
+        with self.assertLogs("prompt_manager.database.health", "ERROR"):
+            report = missing.run_health_checks([str(self.output)], review_count=0)
+        statuses = {c["id"]: c["status"] for c in report["checks"]}
+        self.assertEqual(statuses["integrity"], "failed")
+        self.assertEqual(statuses["missing_files"], "failed")
+        self.assertEqual(statuses["ambiguous_metadata"], "passed")
+        self.assertEqual(report["status"], "incomplete")
+        self.assertFalse(os.path.exists(self.root / "nope.db"))
+
+    def test_integrity_and_foreign_key_issues_reported(self):
+        prompt = self.save("Linked")
+        with self.raw() as conn:
+            conn.execute("CREATE TABLE probe(a)")
+            conn.execute("CREATE INDEX probe_a ON probe(a)")
+            conn.executemany("INSERT INTO probe VALUES(?)", [(i,) for i in range(16)])
+            conn.execute(
+                "INSERT INTO generated_images(prompt_id,image_path,filename,file_path)"
+                " VALUES(?,?,?,?)",
+                (9999, str(self.output / "orphan.png"), "orphan.png", "orphan"),
+            )
+            conn.execute(
+                "INSERT INTO prompt_tags(prompt_id,tag_id) VALUES(?,?)", (prompt, 777)
+            )
+        conn.close()
+        with self.raw() as conn:
+            conn.execute("PRAGMA writable_schema=ON")
+            conn.execute(
+                "UPDATE sqlite_master SET sql='CREATE INDEX probe_a ON probe(a DESC)'"
+                " WHERE name='probe_a'"
+            )
+        conn.close()
+        self.db.close_all()  # the next read sees the edited schema
+        (self.output / "orphan.png").write_bytes(b"png")
+        with self.raw() as conn:
+            expected = [r[0] for r in conn.execute("PRAGMA integrity_check(100)")]
+        conn.close()
+        self.assertGreater(len(expected), 10)
+        report = self.run_checks(review_count=0)
+        integrity = self.check(report, "integrity")
+        self.assertEqual(integrity["status"], "issues")
+        self.assertEqual(integrity["count"], len(expected))
+        self.assertEqual(integrity["detail"], expected[:10])
+        self.assertIn("missing from index probe_a", integrity["detail"][0])
+        foreign = self.check(report, "foreign_keys")
+        self.assertEqual(foreign["status"], "issues")
+        self.assertEqual(foreign["count"], 2)
+        self.assertEqual(
+            sorted(foreign["detail"]),
+            [
+                "generated_images row 1 references a missing prompts row",
+                "prompt_tags row 1 references a missing tags row",
+            ],
+        )
+        consistency = self.check(report, "consistency")
+        self.assertEqual(consistency["status"], "issues")
+        self.assertEqual(consistency["count"], 1)
+        self.assertEqual(
+            consistency["detail"], ["Image 1 references non-existent prompt 9999"]
+        )
+        self.assertEqual(report["status"], "issues")
+
+    def test_canonical_hash_mismatch_counts_rows(self):
+        ids = [self.save(f"Prompt {n}") for n in range(14)]
+        with self.raw() as conn:
+            for n, prompt_id in enumerate(ids[:12]):
+                conn.execute(
+                    "UPDATE prompts SET hash=? WHERE id=?", (f"bad{n}", prompt_id)
+                )
+            conn.execute("UPDATE prompts SET hash=NULL WHERE id=?", (ids[12],))
+        conn.close()
+        # Normalised spellings hash alike: this one is not a mismatch.
+        self.save("  MIXED Case  ")
+        report = self.run_checks(review_count=0)
+        canonical = self.check(report, "canonical_hashes")
+        self.assertEqual(canonical["status"], "issues")
+        self.assertEqual(canonical["count"], 13)
+        self.assertEqual(len(canonical["detail"]), 10)
+        self.assertEqual(
+            canonical["detail"][0],
+            f"Prompt {ids[0]}: the stored hash does not match its text",
+        )
+
+    def test_duplicate_pair_rows_reported(self):
+        pair = self.db.save_prompt_pair("Cat", "")
+        positive = self.db.get_prompt_pair(pair)["positive_prompt_id"]
+        self.db.save_prompt_pair("Dog", "Blur")
+        with self.raw() as conn:
+            # Only the partial index keeps positive-only pairs unique; a restored
+            # or hand-edited database may have lost it.
+            conn.execute("DROP INDEX idx_prompt_pairs_positive_only")
+            conn.executemany(
+                "INSERT INTO prompt_pairs(positive_prompt_id,negative_prompt_id)"
+                " VALUES(?,NULL)",
+                [(positive,), (positive,)],
+            )
+        conn.close()
+        report = self.run_checks(review_count=0)
+        pairs = self.check(report, "pair_uniqueness")
+        self.assertEqual(pairs["status"], "issues")
+        self.assertEqual(pairs["count"], 2)
+        self.assertEqual(
+            pairs["detail"],
+            [f"Positive prompt {positive} with no negative: 3 pair rows"],
+        )
+
+
+class TestMissingFiles(HealthTestCase):
+    def test_missing_files_counted_under_readable_sources_only(self):
+        prompt = self.save("Images")
+        self.image(prompt, "here.png")
+        for n in range(12):
+            self.image(prompt, f"gone-{n}.png", exists=False)
+        elsewhere = self.root / "elsewhere"
+        elsewhere.mkdir()
+        self.image(prompt, "outside.png", exists=False, folder=elsewhere)
+        report = self.run_checks(review_count=0)
+        missing = self.check(report, "missing_files")
+        self.assertEqual(missing["status"], "issues")
+        self.assertEqual(missing["count"], 12)
+        self.assertEqual(len(missing["detail"]), 10)
+        self.assertRegex(missing["detail"][0], r"^Image \d+: gone-0\.png is missing$")
+        self.assertNotIn(str(self.root), " ".join(missing["detail"]))
+        with self.db.model.get_connection() as conn:
+            self.assertEqual(
+                conn.execute("SELECT COUNT(*) FROM generated_images").fetchone()[0],
+                14,
+            )
+
+    def test_links_outside_every_source_are_named_not_counted(self):
+        prompt = self.save("Images")
+        elsewhere = self.root / "elsewhere"
+        elsewhere.mkdir()
+        self.image(prompt, "outside.png", exists=False, folder=elsewhere)
+        missing = self.check(self.run_checks(review_count=0), "missing_files")
+        self.assertEqual(missing["status"], "passed")
+        self.assertEqual(missing["count"], 0)
+        self.assertEqual(
+            missing["detail"],
+            ["1 link outside the image sources was not checked."],
+        )
+
+    def test_missing_file_vs_unavailable_root(self):
+        prompt = self.save("Images")
+        absent = self.root / "unmounted"
+        absent.mkdir()
+        self.image(prompt, "on-nas.png", folder=absent)
+        self.image(prompt, "gone.png", exists=False)
+        os.unlink(absent / "on-nas.png")
+        absent.rmdir()
+        report = self.run_checks([str(self.output), str(absent)], review_count=0)
+        missing = self.check(report, "missing_files")
+        self.assertEqual(missing["status"], "unavailable")
+        self.assertIsNone(missing["count"])
+        self.assertEqual(
+            missing["detail"],
+            ["1 image source is unavailable, so missing files were not counted."],
+        )
+        self.assertEqual(report["status"], "incomplete")
+
+    def test_no_sources_is_unavailable(self):
+        self.save("Images")
+        missing = self.check(self.run_checks([], review_count=0), "missing_files")
+        self.assertEqual(missing["status"], "unavailable")
+        self.assertIsNone(missing["count"])
+        self.assertEqual(missing["detail"], ["No image source is available."])
+
+
+class TestAmbiguousMetadata(HealthTestCase):
+    def test_ambiguous_metadata_unavailable_without_review_queue(self):
+        self.save("Any")
+        report = self.run_checks()
+        ambiguous = self.check(report, "ambiguous_metadata")
+        self.assertEqual(ambiguous["status"], "unavailable")
+        self.assertIsNone(ambiguous["count"])
+        self.assertEqual(ambiguous["detail"], ["Needs review is not available yet."])
+        self.assertEqual(report["status"], "incomplete")
+
+    def test_review_count_reports_images_needing_review(self):
+        report = self.run_checks(review_count=3)
+        ambiguous = self.check(report, "ambiguous_metadata")
+        self.assertEqual(ambiguous["status"], "issues")
+        self.assertEqual(ambiguous["count"], 3)
+        self.assertEqual(ambiguous["detail"], [])
+        self.assertEqual(report["status"], "issues")
+
+    def test_invalid_review_count_is_rejected(self):
+        for value in (-1, True, "3", 1.5):
+            with self.assertRaises(ValueError):
+                self.run_checks(review_count=value)
+
+
+class TestOverallStatus(unittest.TestCase):
+    def test_overall_status_precedence(self):
+        def checks(*statuses):
+            return [{"status": status} for status in statuses]
+
+        self.assertEqual(health.overall_status(checks("passed", "passed")), "passed")
+        self.assertEqual(health.overall_status(checks("passed", "issues")), "issues")
+        self.assertEqual(
+            health.overall_status(checks("issues", "unavailable")), "incomplete"
+        )
+        self.assertEqual(health.overall_status(checks("failed")), "incomplete")
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_library.py

@@ -1706,6 +1706,82 @@ class TestLibrary(unittest.TestCase):
             self.db.list_library_images(positive, pair_id=foreign, archive="all")
 
 
+class LibraryCounts(unittest.TestCase):
+    """System › Overview counts and usage statistics follow Library eligibility."""
+
+    setUp = TestLibrary.setUp
+    tearDown = TestLibrary.tearDown
+    save = TestLibrary.save
+    state = TestLibrary.state
+
+    def test_counts_use_library_eligibility(self):
+        self.db.save_prompt_pair("Cat", "Blur")
+        archived = self.db.save_prompt_pair("Cat", "Noise")
+        self.db.archive_prompt_pair(archived)
+        titled_pair = self.db.save_prompt_pair("Dog", "Titled negative")
+        titled = self.db.get_prompt_pair(titled_pair)["negative_prompt_id"]
+        self.state(titled, "A titled parent")  # zero pairs of its own: counted
+        legacy = self.save("Legacy")
+        self.db.link_image_to_prompt(legacy, str(Path(self.temp.name) / "a.png"))
+        self.db.link_image_to_prompt(legacy, str(Path(self.temp.name) / "b.png"))
+        counts = self.db.library_counts()
+        # Cat, Dog, the titled negative and Legacy; "Blur" and "Noise" are
+        # negative-only.
+        self.assertEqual(counts["positive_prompts"], 4)
+        # Cat+Blur and Dog+Titled negative; the archived pair is excluded.
+        self.assertEqual(counts["negative_variants"], 2)
+        self.assertEqual(counts["images"], 2)
+        listed = self.db.list_library()
+        self.assertEqual(counts["positive_prompts"], listed["total"])
+        self.assertEqual(counts["negative_variants"], listed["variant_total"])
+
+    def test_counts_of_an_empty_library_are_zero(self):
+        self.assertEqual(
+            self.db.library_counts(),
+            {"positive_prompts": 0, "negative_variants": 0, "images": 0},
+        )
+
+    def test_statistics_sum_real_run_counts(self):
+        a = self.save("Alpha", category="Landscapes", rating=5)
+        b = self.save("Beta", category="Landscapes", rating=4)
+        self.save("Gamma", category="Landscapes")  # unrated: not in the average
+        c = self.save("Delta", category="Portraits", rating=3)
+        blank = self.save("Epsilon", category="  ")
+        uncategorised = self.save("Zeta")
+        self.db.record_prompt_use(a, times=3)
+        self.db.record_prompt_use(b, times=2)
+        self.db.record_prompt_use(c)
+        self.db.record_prompt_use(blank)
+        self.db.record_prompt_use(uncategorised, times=4)
+        # A negative-only prompt's runs are the pair's runs: never counted twice.
+        pair = self.db.save_prompt_pair("Alpha", "Only negative")
+        negative = self.db.get_prompt_pair(pair)["negative_prompt_id"]
+        with self.db.model.get_connection() as connection:
+            connection.execute(
+                "UPDATE prompts SET category='Landscapes',run_count=50 WHERE id=?",
+                (negative,),
+            )
+        rows = self.db.usage_by_category()
+        self.assertEqual(
+            rows,
+            [
+                {"category": "Landscapes", "runs": 5, "average_rating": 4.5},
+                {"category": None, "runs": 5, "average_rating": None},
+                {"category": "Portraits", "runs": 1, "average_rating": 3.0},
+            ],
+        )
+
+    def test_counts_and_usage_write_nothing(self):
+        self.save("Alpha", category="A")
+        with self.db.model.get_connection() as connection:
+            before = connection.total_changes
+        self.db.library_counts()
+        self.db.usage_by_category()
+        with self.db.model.get_connection() as connection:
+            self.assertEqual(connection.total_changes, before)
+            self.assertFalse(connection.in_transaction)
+
+
 class DisplayTitleUnits(unittest.TestCase):
     """The first-line title is cut like the Library's JavaScript title()."""
 

tests/test_system_api.py

@@ -0,0 +1,304 @@
+"""System routes (py/api/system.py): health check job, Overview and statistics.
+
+The health check runs off-loop as a ``read`` registry job of kind ``health``
+(one at a time, beside a scan), and the API keeps its last report. The
+Overview reads only real values: Library-eligible counts, indexed images, the
+last health result or null, no backup until one exists, the effective sources
+and whether they were auto-detected, and the thumbnail plan's missing count.
+"""
+
+import asyncio
+import os
+import sys
+import threading
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+_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.jobs import HEALTH_JOB_KIND  # noqa: E402
+from py.config import GalleryConfig  # noqa: E402
+from utils.hashing import generate_prompt_hash  # noqa: E402
+
+try:
+    from tests.test_api_admin import AdminAPITestCase  # noqa: E402
+except ImportError:  # discovered with tests/ as the top-level directory
+    from test_api_admin import AdminAPITestCase  # noqa: E402
+
+WAIT = 5
+
+
+class SystemAPITestCase(AdminAPITestCase):
+    async def get_application(self):
+        app = await super().get_application()
+        # Auto-detection searches the disk; here it finds the fake ComfyUI only.
+        self.api._find_comfyui_output_dir = lambda: (
+            str(self.output_dir) if self.output_dir.is_dir() else None
+        )
+        return app
+
+    def save(self, text, **metadata):
+        return self.api.db.save_prompt(
+            text, prompt_hash=generate_prompt_hash(text), **metadata
+        )
+
+    def image(self, prompt_id, name, *, exists=True, folder=None):
+        path = Path(folder or self.output_dir) / name
+        if exists:
+            path.write_bytes(b"png")
+        return self.api.db.link_image_to_prompt(prompt_id, str(path))
+
+    async def json(self, method, url, status=200):
+        response = await self.client.request(method, url)
+        self.assertEqual(response.status, status, await response.text())
+        return await response.json()
+
+    async def run_health(self):
+        body = await self.json("POST", "/prompt_manager/system/health", 202)
+        job = self.api._jobs.get(body["job"]["id"])
+        await asyncio.wait_for(asyncio.shield(job.task), WAIT)
+        return body["job"], job
+
+
+class TestHealthRoute(SystemAPITestCase):
+    async def test_no_report_before_a_check(self):
+        body = await self.json("GET", "/prompt_manager/system/health")
+        self.assertEqual(body, {"success": True, "report": None})
+
+    async def test_health_check_runs_as_a_read_job_and_keeps_its_report(self):
+        prompt = self.save("Images")
+        self.image(prompt, "kept.png")
+        self.image(prompt, "gone.png", exists=False)
+        summary, job = await self.run_health()
+        self.assertEqual(summary["kind"], HEALTH_JOB_KIND)
+        self.assertEqual(summary["title"], "Health check")
+        self.assertEqual(summary["admission"], "read")
+        self.assertFalse(summary["cancellable"])
+        self.assertEqual(summary["counters"], {})
+        self.assertEqual(job.state, "finished")
+        self.assertEqual(job.progress, 100)
+        self.assertEqual(job.status_line, "Health check incomplete")
+        report = (await self.json("GET", "/prompt_manager/system/health"))["report"]
+        self.assertEqual(report["status"], "incomplete")
+        missing = next(c for c in report["checks"] if c["id"] == "missing_files")
+        self.assertEqual((missing["status"], missing["count"]), ("issues", 1))
+        ambiguous = next(c for c in report["checks"] if c["id"] == "ambiguous_metadata")
+        self.assertEqual(ambiguous["status"], "unavailable")
+        self.assertNotIn(self.tmpdir, str(report))
+
+    async def test_health_check_is_single_flight_and_runs_beside_a_scan(self):
+        release = threading.Event()
+        entered = threading.Event()
+        real = self.api.db.run_health_checks
+
+        def gated(*args, **kwargs):
+            entered.set()
+            release.wait(WAIT)
+            return real(*args, **kwargs)
+
+        scan_gate = asyncio.Event()
+
+        async def scanning(job):
+            await scan_gate.wait()
+
+        scan = self.api._jobs.start(
+            "scan", "Metadata scan", admission="records", runner=scanning
+        )
+        with patch.object(self.api.db, "run_health_checks", side_effect=gated):
+            first = await self.json("POST", "/prompt_manager/system/health", 202)
+            loop = asyncio.get_running_loop()
+            await loop.run_in_executor(None, entered.wait, WAIT)
+            busy = await self.json("POST", "/prompt_manager/system/health", 409)
+            self.assertEqual(busy["code"], "busy")
+            self.assertEqual(
+                busy["running"], {"id": first["job"]["id"], "title": "Health check"}
+            )
+            release.set()
+            await asyncio.wait_for(self.api._jobs.get(first["job"]["id"]).task, WAIT)
+        scan_gate.set()
+        await asyncio.wait_for(scan.task, WAIT)
+
+    async def test_failed_check_keeps_the_previous_report(self):
+        await self.run_health()
+        before = (await self.json("GET", "/prompt_manager/system/health"))["report"]
+        with (
+            patch.object(
+                self.api.db, "run_health_checks", side_effect=RuntimeError("boom")
+            ),
+            self.assertLogs("prompt_manager.api", "ERROR"),
+        ):
+            body = await self.json("POST", "/prompt_manager/system/health", 202)
+            job = self.api._jobs.get(body["job"]["id"])
+            await asyncio.wait_for(asyncio.shield(job.task), WAIT)
+        self.assertEqual(job.state, "failed")
+        after = (await self.json("GET", "/prompt_manager/system/health"))["report"]
+        self.assertEqual(after, before)
+
+    async def test_absent_configured_source_makes_missing_files_unavailable(self):
+        prompt = self.save("Images")
+        self.image(prompt, "gone.png", exists=False)
+        absent = self.comfy_dir / "output" / "unmounted"
+        GalleryConfig.MONITORING_DIRECTORIES = [
+            self._root(self.output_dir),
+            str(absent),
+        ]
+        await self.run_health()
+        report = (await self.json("GET", "/prompt_manager/system/health"))["report"]
+        missing = next(c for c in report["checks"] if c["id"] == "missing_files")
+        self.assertEqual(missing["status"], "unavailable")
+        self.assertIsNone(missing["count"])
+
+
+class TestOverviewRoute(SystemAPITestCase):
+    async def test_overview_reports_real_counts(self):
+        self.api.db.save_prompt_pair("Cat", "Blur")
+        archived = self.api.db.save_prompt_pair("Cat", "Noise")
+        self.api.db.archive_prompt_pair(archived)
+        legacy = self.save("Legacy")
+        self.image(legacy, "a.png")
+        self.image(legacy, "b.png")
+        body = await self.json("GET", "/prompt_manager/system/overview")
+        self.assertTrue(body["success"])
+        self.assertEqual(
+            body["library"], {"positive_prompts": 2, "negative_variants": 1}
+        )
+        self.assertEqual(body["images"], {"indexed": 2})
+        self.assertEqual(
+            set(body),
+            {
+                "success",
+                "library",
+                "images",
+                "health",
+                "backup",
+                "needs_review",
+                "sources",
+                "thumbnails",
+            },
+        )
+
+    async def test_overview_never_fabricates_backup_or_health(self):
+        body = await self.json("GET", "/prompt_manager/system/overview")
+        self.assertIsNone(body["health"])
+        self.assertIsNone(body["backup"])
+        self.assertEqual(body["needs_review"], {"pending": None})
+        self.save("Any")
+        await self.run_health()
+        report = (await self.json("GET", "/prompt_manager/system/health"))["report"]
+        body = await self.json("GET", "/prompt_manager/system/overview")
+        self.assertEqual(
+            body["health"],
+            {"checked_at": report["checked_at"], "status": "incomplete", "issues": 0},
+        )
+        self.assertIsNone(body["backup"])
+
+    async def test_overview_health_counts_checks_with_issues(self):
+        prompt = self.save("Images")
+        self.image(prompt, "gone.png", exists=False)
+        self.image(prompt, "gone-too.png", exists=False)
+        await self.run_health()
+        body = await self.json("GET", "/prompt_manager/system/overview")
+        self.assertEqual(body["health"]["issues"], 1)
+
+    async def test_overview_sources_distinguish_auto_detect(self):
+        body = await self.json("GET", "/prompt_manager/system/overview")
+        self.assertEqual(
+            body["sources"], {"configured": 0, "effective": 1, "auto_detected": True}
+        )
+        second = self.output_dir / "archive"
+        second.mkdir()
+        GalleryConfig.MONITORING_DIRECTORIES = [
+            self._root(self.output_dir),
+            self._root(second),
+        ]
+        body = await self.json("GET", "/prompt_manager/system/overview")
+        self.assertEqual(
+            body["sources"], {"configured": 2, "effective": 2, "auto_detected": False}
+        )
+        # Every configured source unavailable: ComfyUI's output is used instead.
+        GalleryConfig.MONITORING_DIRECTORIES = [str(self.comfy_dir / "absent")]
+        body = await self.json("GET", "/prompt_manager/system/overview")
+        self.assertEqual(
+            body["sources"], {"configured": 1, "effective": 1, "auto_detected": True}
+        )
+
+    async def test_overview_thumbnails_follow_the_status_plan(self):
+        (self.output_dir / "one.png").write_bytes(b"png")
+        (self.output_dir / "two.png").write_bytes(b"png")
+        status = await self.json("GET", "/prompt_manager/images/thumbnails/status")
+        body = await self.json("GET", "/prompt_manager/system/overview")
+        self.assertEqual(body["thumbnails"], {"missing": status["pending"]})
+        self.assertEqual(body["thumbnails"]["missing"], 2)
+
+    async def test_overview_thumbnails_unknown_without_sources(self):
+        self.output_dir.rmdir()
+        body = await self.json("GET", "/prompt_manager/system/overview")
+        self.assertEqual(body["thumbnails"], {"missing": None})
+        self.assertEqual(
+            body["sources"], {"configured": 0, "effective": 0, "auto_detected": False}
+        )
+
+    async def test_overview_thumbnail_failure_is_unknown_not_zero(self):
+        with (
+            patch.object(
+                self.api, "_plan_gallery_thumbnails", side_effect=OSError("denied")
+            ),
+            self.assertLogs("prompt_manager.api", "WARNING"),
+        ):
+            body = await self.json("GET", "/prompt_manager/system/overview")
+        self.assertTrue(body["success"])
+        self.assertEqual(body["thumbnails"], {"missing": None})
+
+    async def test_overview_failure_is_an_error_envelope(self):
+        with (
+            patch.object(
+                self.api.db, "library_counts", side_effect=RuntimeError("secret")
+            ),
+            self.assertLogs("prompt_manager.api", "ERROR"),
+        ):
+            body = await self.json("GET", "/prompt_manager/system/overview", 500)
+        self.assertEqual(
+            body,
+            {"success": False, "code": None, "error": "Failed to read the overview"},
+        )
+
+
+class TestStatisticsRoute(SystemAPITestCase):
+    async def test_statistics_sum_real_run_counts(self):
+        a = self.save("Alpha", category="Landscapes", rating=5)
+        b = self.save("Beta", category="Portraits", rating=2)
+        self.api.db.record_prompt_use(a, times=3)
+        self.api.db.record_prompt_use(b)
+        body = await self.json("GET", "/prompt_manager/system/statistics")
+        self.assertEqual(
+            body["by_category"],
+            [
+                {"category": "Landscapes", "runs": 3, "average_rating": 5.0},
+                {"category": "Portraits", "runs": 1, "average_rating": 2.0},
+            ],
+        )
+        self.assertEqual(body["totals"], self.api.db.get_statistics())
+        self.assertTrue(body["success"])
+
+    async def test_statistics_failure_is_an_error_envelope(self):
+        with (
+            patch.object(
+                self.api.db, "usage_by_category", side_effect=RuntimeError("secret")
+            ),
+            self.assertLogs("prompt_manager.api", "ERROR"),
+        ):
+            body = await self.json("GET", "/prompt_manager/system/statistics", 500)
+        self.assertEqual(
+            body,
+            {"success": False, "code": None, "error": "Failed to read statistics"},
+        )
+
+
+if __name__ == "__main__":
+    import unittest
+
+    unittest.main()