perf(maintenance): evaluate each unused-prompt protection once (#586)

Duplicate prompts, unused components, missing links, workflow storage and
rotated logs now record one protected row per kept item, in the review's
wording, and the job report lists every one of them. Kept links and log
files keep their path. The duplicate file cleanup records each copy the
review protected in its Kept group.

database/health.py

@@ -199,6 +199,23 @@ def link_states(rows, roots, readable):
     check's ``missing_files`` and the maintenance missing-link plan both
     judge links this way, so their counts agree.
     """
+    for row, owner, state in _placed_links(rows, roots, readable):
+        if state is None:
+            state = "present" if os.path.exists(row[1]) else "missing"
+        yield row, owner, state
+
+
+def held_links(rows, roots, readable):
+    """``(row, owner, state)`` for each link ``link_states`` holds back
+    (``unavailable`` or ``outside``), without looking at any file (#586)."""
+    for row, owner, state in _placed_links(rows, roots, readable):
+        if state is not None:
+            yield row, owner, state
+
+
+def _placed_links(rows, roots, readable):
+    """``link_states`` before any file is looked at: state None means the
+    link lies under a readable source."""
     prefixes = [_root_prefix(root) for root in roots]
     for row in rows:
         owner = _image_owner(row[1], row[3], prefixes)
@@ -206,10 +223,8 @@ def link_states(rows, roots, readable):
             yield row, None, "outside"
         elif not readable[owner]:
             yield row, owner, "unavailable"
-        elif os.path.exists(row[1]):
-            yield row, owner, "present"
         else:
-            yield row, owner, "missing"
+            yield row, owner, None
 
 
 def _unavailable_line(unavailable, *, partial):

database/maintenance.py

@@ -43,7 +43,7 @@ from dataclasses import dataclass, field
 from pathlib import Path
 
 from .execution_contexts import LIVE_HINT_CTES
-from .health import BATCH_ROWS, link_states, readable_roots, readonly_uri
+from .health import BATCH_ROWS, held_links, link_states, readable_roots, readonly_uri
 from .image_removal_schema import TABLE as MARKS_TABLE
 from .image_removal_schema import clear_marks, has_marks_table
 from .pairs import PairDependencyError
@@ -130,6 +130,16 @@ _UNLINKED_PROTECTIONS_SQL = (
     + "), 0) FROM prompts p WHERE NOT EXISTS "
     "(SELECT 1 FROM generated_images gi WHERE gi.prompt_id = p.id)"
 )
+# Each prompt no image links to, with which protections keep it (#586); the
+# ones with none are filtered out in Python, so each predicate runs once.
+_UNLINKED_KEPT_SQL = (
+    "WITH "
+    + LIVE_HINT_CTES
+    + " SELECT p.id, "
+    + ", ".join(f"({predicate})" for predicate, _many, _one in PROTECTIONS)
+    + " FROM prompts p WHERE NOT EXISTS "
+    "(SELECT 1 FROM generated_images gi WHERE gi.prompt_id = p.id) ORDER BY p.id"
+)
 
 # Duplicate prompt groups, as the legacy cleanup finds them (case-insensitive
 # text; GROUP_CONCAT order is sorted by created_at afterwards).
@@ -159,6 +169,14 @@ _PROMPT_STATE_SQL = (
     "ORDER BY id",
 )
 
+# Why the missing-link cleanup holds a link back: (one, many) (#586).
+HELD_LINK_REASONS = {
+    "unavailable": (
+        "link under an unavailable source",
+        "links under unavailable sources",
+    ),
+    "outside": ("link outside the image sources", "links outside the image sources"),
+}
 NO_FILES_DELETED = "No image file is deleted."
 LINKS_ONLY = "Database links only; no image file is deleted."
 UNUSED_DETAIL = "No image, saved pair, Library state or recorded run uses it"
@@ -172,6 +190,7 @@ CONFLICT_REASONS = {
         "groups have their own Library titles",
     ),
 }
+PAIRED_DETAIL = "Kept: belongs to a saved pair"
 CONFLICT_DETAILS = {
     "collision": "Kept: two copies link the same image",
     "title": "Kept: a copy has its own Library title",
@@ -413,6 +432,35 @@ def _source_name(root, index):
     return os.path.basename(str(root).rstrip("\\/")) or f"Source {index + 1}"
 
 
+def _kept_duplicates(paired, conflicted):
+    """``{prompt_id, reasons}`` for each prompt of a group kept whole (#586).
+
+    The reasons are the review's details without "Kept: ".
+    """
+    held = [(ids, PAIRED_DETAIL) for ids in paired] + [
+        (ids, CONFLICT_DETAILS[reason]) for ids, reason in conflicted
+    ]
+    return [
+        {"prompt_id": prompt_id, "reasons": [detail.removeprefix("Kept: ")]}
+        for ids, detail in held
+        for prompt_id in ids
+    ]
+
+
+def _kept_unused(conn):
+    """``{prompt_id, reasons}`` for each prompt the unused prune keeps (#586).
+
+    The reasons are the review's, one per protection that applies, in the
+    order the plan names them.
+    """
+    kept = []
+    for row in conn.execute(_UNLINKED_KEPT_SQL):
+        reasons = [one for flag, (_p, _m, one) in zip(row[1:], PROTECTIONS) if flag]
+        if reasons:
+            kept.append({"prompt_id": row[0], "reasons": reasons})
+    return kept
+
+
 class MaintenanceOperations:
     """Mixin for PromptDatabase: maintenance previews and their apply."""
 
@@ -592,7 +640,7 @@ class MaintenanceOperations:
             + [
                 {
                     "label": _excerpt(texts.get(group[0])),
-                    "detail": "Kept: belongs to a saved pair",
+                    "detail": PAIRED_DETAIL,
                 }
                 for group in paired
             ]
@@ -670,6 +718,24 @@ class MaintenanceOperations:
             ],
         )
 
+    def kept_links(self, roots):
+        """``{image_path, owner, reason}`` for each link the missing-link
+        cleanup holds back under ``roots`` (blocking, reads only, no file
+        opened); ``owner`` indexes ``roots`` (None outside every source).
+
+        ``reason`` is the review's (``HELD_LINK_REASONS``), one link each.
+        """
+        roots = [str(root) for root in roots]
+        rows = _read_snapshot(self.model.db_path, self._link_rows)
+        return [
+            {
+                "image_path": row[1],
+                "owner": owner,
+                "reason": HELD_LINK_REASONS[state][0],
+            }
+            for row, owner, state in held_links(rows, roots, readable_roots(roots))
+        ]
+
     @staticmethod
     def _links_fingerprint(candidates, stale=()):
         fingerprint = _Fingerprint("missing_links")
@@ -689,23 +755,11 @@ class MaintenanceOperations:
                 unavailable += 1
             elif state == "outside":
                 outside += 1
-        reasons = []
-        if unavailable:
-            reasons.append(
-                _plural(
-                    unavailable,
-                    "link under an unavailable source",
-                    "links under unavailable sources",
-                )
-            )
-        if outside:
-            reasons.append(
-                _plural(
-                    outside,
-                    "link outside the image sources",
-                    "links outside the image sources",
-                )
-            )
+        reasons = [
+            _plural(count, *HELD_LINK_REASONS[state])
+            for state, count in (("unavailable", unavailable), ("outside", outside))
+            if count
+        ]
         count = len(candidates)
         parts = []
         if count:
@@ -809,7 +863,10 @@ class MaintenanceOperations:
         (``{prompt_id[, kept_id]}`` or ``{image_path}``). ``marks`` are a
         missing-links plan's stale removal marks; that apply also returns
         ``marks``, how many it forgot. ``pair_metadata`` returns every filled
-        prompt, uncapped (its job keeps the complete record, #550).
+        prompt, uncapped (its job keeps the complete record, #550). The
+        prompt operations also return ``kept``: every prompt the plan keeps,
+        ``{prompt_id, reasons}`` in the review's wording (#550, #586); the
+        links a missing-links plan keeps are read by ``kept_links``.
         """
         if operation not in MAINTENANCE_OPERATIONS or operation == "workflow_storage":
             raise ValueError(f"Unknown maintenance operation {operation!r}")
@@ -835,10 +892,11 @@ class MaintenanceOperations:
             return appliers[operation](conn, fingerprint, targets, roots)
 
     def _apply_duplicate_prompts(self, conn, fingerprint, targets, _roots):
-        groups, _paired, _conflicted, current = self._duplicates_fingerprint(conn)
+        groups, paired, conflicted, current = self._duplicates_fingerprint(conn)
         if current != fingerprint or groups != targets:
             raise PlanChanged("duplicate_prompts")
         changed, rows = 0, []
+        kept = _kept_duplicates(paired, conflicted)
         for sorted_ids in groups:
             removed, _moved = self._merge_duplicate_group(conn, sorted_ids, strict=True)
             changed += removed
@@ -846,7 +904,7 @@ class MaintenanceOperations:
                 {"prompt_id": duplicate_id, "kept_id": sorted_ids[0]}
                 for duplicate_id in sorted_ids[1:]
             )
-        return {"changed": changed, "rows": rows[:MAX_RESULT_ROWS]}
+        return {"changed": changed, "rows": rows[:MAX_RESULT_ROWS], "kept": kept}
 
     def _apply_unused_components(self, conn, fingerprint, targets, _roots):
         ids, current = self._unused_fingerprint(conn)
@@ -858,7 +916,7 @@ class MaintenanceOperations:
             {"prompt_id": prompt_id, "text": _excerpt(texts.get(prompt_id))}
             for prompt_id in ids[:MAX_RESULT_ROWS]
         ]
-        return {"changed": changed, "rows": rows}
+        return {"changed": changed, "rows": rows, "kept": _kept_unused(conn)}
 
     def _apply_missing_links(self, conn, fingerprint, targets, roots, marks):
         rows = []

database/workflow_storage.py

@@ -112,20 +112,22 @@ QUERY_PLAN_STATEMENTS = {
 
 
 def _rewrite_row(conn, image_id, row, verified, checked):
-    """Store the row's inline texts as documents; returns (changed, protected, size).
+    """Store the row's inline texts as documents; returns (changed, kept, size).
+
+    ``kept`` names the columns whose value stays inline (not text).
 
     ``verified`` maps document ids already read back in this batch to their
     text, and ``checked`` holds the reused ids ``store_text`` already
     checked, so a document shared by many rows is decompressed once.
     """
-    changes, replaced = {}, []
-    protected = size = 0
+    changes, replaced, kept = {}, [], []
+    size = 0
     for index, (column, reference) in enumerate(TEXT_COLUMNS):
         value, current = row[index], row[index + len(TEXT_COLUMNS)]
         if value is None:
             continue
         if not storable(value):
-            protected += 1
+            kept.append(column)
             continue
         document_id = store_text(conn, value, checked)
         if document_id not in verified:
@@ -143,7 +145,7 @@ def _rewrite_row(conn, image_id, row, verified, checked):
             [*changes.values(), image_id],
         )
     release_documents(conn, replaced)
-    return bool(changes), protected, size
+    return bool(changes), kept, size
 
 
 def migrate_batch(
@@ -160,9 +162,10 @@ def migrate_batch(
     savepoint: a row that cannot be rewritten (``_row_error``) is rolled
     back alone, stays inline and is listed in ``errors``; anything else rolls
     the whole batch back and propagates. Returns ``{last_id, rows,
-    protected, errors, done}``: the last row looked at, rows rewritten,
-    inline values kept (not text), the ids kept by an error and whether
-    there was nothing left to look at.
+    protected, kept, errors, done}``: the last row looked at, rows
+    rewritten, inline values kept (not text) and each of them as ``(image
+    id, column)`` (#586), the ids kept by an error and whether there was
+    nothing left to look at.
     """
     if conn.in_transaction:
         raise RuntimeError("The migration needs its own transaction")
@@ -174,6 +177,7 @@ def migrate_batch(
             "last_id": after_id,
             "rows": 0,
             "protected": 0,
+            "kept": [],
             "errors": [],
             "done": not ids,
         }
@@ -182,7 +186,7 @@ def migrate_batch(
             conn.execute("SAVEPOINT workflow_row")
             try:
                 row = conn.execute(ROW_SQL, (image_id,)).fetchone()
-                changed, protected, row_size = _rewrite_row(
+                changed, kept, row_size = _rewrite_row(
                     conn, image_id, row, verified, checked
                 )
             except Exception as error:
@@ -194,11 +198,12 @@ def migrate_batch(
                 checked.clear()
                 logger.warning(f"Image record {image_id} kept inline: {error}")
                 result["errors"].append(image_id)
-                changed, protected, row_size = False, 0, 0
+                changed, kept, row_size = False, [], 0
             else:
                 conn.execute("RELEASE workflow_row")
             result["rows"] += int(changed)
-            result["protected"] += protected
+            result["protected"] += len(kept)
+            result["kept"].extend((image_id, column) for column in kept)
             result["last_id"] = image_id
             size += row_size
             if size >= max_bytes or _clock() - started >= max_seconds:

py/api/duplicate_files.py

@@ -65,6 +65,7 @@ from .duplicate_review import (  # noqa: F401 (CHANGED, OUTSIDE_SOURCE: public)
     NOT_ATTEMPTED,
     OUTSIDE_SOURCE,
     RECORDS_FAILED,
+    REVIEW_KEPT_DETAILS,
     ROW_DETAILS,
     SAME_FILE,
     SOURCES_CHANGED,
@@ -434,6 +435,7 @@ class DuplicateFilesMixin:
             {
                 "targets": review["targets"],
                 "kept": review["kept"],
+                "held": review["held"],
                 "protected": review["public"]["protected"]["count"],
             },
         )
@@ -446,7 +448,7 @@ class DuplicateFilesMixin:
         job hashes each file again right before it unlinks it.
         """
         listed = self._listed_sources()
-        targets, kept, protected = [], [], {}
+        targets, kept, protected, held = [], [], {}, []
         for group, keepers in selection.values():
             removals = [r for r in group["files"] if r not in keepers]
             for record in removals:
@@ -467,6 +469,7 @@ class DuplicateFilesMixin:
                     problem = _bound_problem(record, group, keeper)
                 if problem:
                     protected[problem] = protected.get(problem, 0) + 1
+                    held.append(self._copy_row(record, REVIEW_KEPT_DETAILS[problem]))
                     continue
                 reviewed.append(self._target(group, record, keeper))
             if reviewed:
@@ -488,6 +491,7 @@ class DuplicateFilesMixin:
         return {
             "targets": targets,
             "kept": kept,
+            "held": held,
             "fingerprint": _fingerprint(targets),
             "public": _public_plan(targets, protected, records),
         }
@@ -584,6 +588,13 @@ class DuplicateFilesMixin:
 
     async def _remove_duplicates(self, job, db, plan, backup):
         targets = plan.payload["targets"]
+        # #586: every copy the review protected, with its reason, and the
+        # whole group in the report.
+        held = plan.payload["held"]
+        job.record_counted("kept", len(held), held, complete=True)
+        if held:
+            copies = _plural(len(held), "copy", "copies")
+            job.log_line(f"Kept {copies} the review protected")
         for row in plan.payload["kept"]:
             job.record("kept", row, f"= {row['root']}/{row['path']} · {KEPT_COPY}")
         job.update(total=len(targets), status_line="Removing duplicate files")

py/api/duplicate_review.py

@@ -70,6 +70,18 @@ PROTECTED_REASONS = {
         "files kept: their kept copy changed or is gone",
     ),
 }
+# The job's row for a copy the review protected (#586): the review's reason.
+REVIEW_KEPT_DETAILS = {
+    MAYBE_SAME: "Kept: may be the same file reached through another mount or share",
+    NO_FILE_ID: "Kept: can't confirm this is a separate file on this drive",
+    SAME: "Kept: the same file as its kept copy",
+    OUTSIDE_SOURCE: "Kept: not inside a configured source",
+    LINK: "Kept: a link, not a regular file",
+    CHANGED_CONTENT: "Kept: changed since the search",
+    GONE: "Kept: no longer exists",
+    UNREADABLE: "Kept: could not be read",
+    KEEPER: "Kept: its kept copy changed or is gone",
+}
 # Job outcome rows (never an OS error text).
 CHANGED = "The file changed since the review; it was kept."
 KEEPER_CHANGED = "The kept copy changed or is gone; this file was kept."

py/api/jobs.py

@@ -288,15 +288,16 @@ class Job:
         index = row.get("root_index")
         if isinstance(index, bool) or not isinstance(index, int) or index < 0:
             index = None
+        shaped = {
+            "path": path,
+            "root": str(row.get("root") or ""),
+            "root_index": index,
+            "detail": str(row.get("detail") or ""),
+        }
+        if group in self.complete:  # a group kept whole for the report (#586)
+            self.complete[group].append(dict(shaped))
         if len(rows) < MAX_OUTCOME_ROWS:
-            rows.append(
-                {
-                    "path": path,
-                    "root": str(row.get("root") or ""),
-                    "root_index": index,
-                    "detail": str(row.get("detail") or ""),
-                }
-            )
+            rows.append(shaped)
         self.log_line(log_line)
 
     def record_counted(self, group, count, rows=(), *, complete=False):

py/api/log_cleanup.py

@@ -68,6 +68,23 @@ def _count(count, singular, plural):
     return f"{count} {singular if count == 1 else plural}"
 
 
+# Each kept entry's row in the job (#586), in the review's "Kept: ..." words.
+KEPT_DETAILS = {
+    "active": "Kept: the active log",
+    "link": "Kept: a link",
+    "other": "Kept: another file",
+}
+
+
+def _kept_files(rows):
+    """``{name, kind}`` of each entry the cleanup keeps, in folder order."""
+    return [
+        {"name": row["name"], "kind": row["kind"]}
+        for row in rows
+        if row["kind"] in KEPT_DETAILS
+    ]
+
+
 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")
@@ -129,7 +146,7 @@ class LogCleanupMixin:
             token = store.issue(
                 ROTATED_LOGS,
                 json.dumps(files, sort_keys=True),
-                {"files": files, "protected": protected},
+                {"files": files, "protected": protected, "kept": _kept_files(rows)},
                 (),
             )
         except ScopeTooLarge:
@@ -198,7 +215,11 @@ class LogCleanupMixin:
             job.record(
                 "errors", {"path": name, "detail": detail}, f"! {name} · {detail}"
             )
-        job.record_counted("protected", plan.payload["protected"])
+        kept = [
+            {"path": row["name"], "detail": KEPT_DETAILS[row["kind"]]}
+            for row in plan.payload["kept"]
+        ]
+        job.record_counted("protected", len(kept), kept, complete=True)
         removed, failed = len(result["removed"]), len(result["failed"])
         line = f"Deleted {_count(removed, 'rotated log copy', 'rotated log copies')}"
         if failed:

py/api/maintenance.py

@@ -697,19 +697,13 @@ class MaintenanceRoutesMixin:
             rows,
             complete=plan.operation in COMPLETE_RECORD,
         )
-        if "kept" in result:
-            # #550: each kept record with its reason, like the changed ones.
-            kept = [
-                {
-                    "path": "",
-                    "detail": f"Prompt {item['prompt_id']}: kept, "
-                    + "; ".join(item["reasons"]),
-                }
-                for item in result["kept"]
-            ]
-            job.record_counted("protected", len(kept), kept, complete=True)
+        kept = await self._run_in_executor(self._kept_rows, db, plan, result)
+        if kept is None:
+            job.record_counted("protected", payload["protected"])  # none kept
         else:
-            job.record_counted("protected", payload["protected"])
+            # #550, #586: each kept record with its reason, like the changed
+            # ones, and every one of them in the report.
+            job.record_counted("protected", len(kept), kept, complete=True)
         line = _changed_line(
             plan.operation, result["changed"], result.get("marks", 0), result
         )
@@ -717,6 +711,47 @@ class MaintenanceRoutesMixin:
         job.update(progress=100, status_line=f"{line} · {backup_note(backup)}")
         return None
 
+    def _kept_rows(self, db, plan, result):
+        """Outcome rows for the records the apply kept (blocking), or None.
+
+        Prompts are records (no path); a missing-links plan's kept links
+        name their file, relative to its output root, so Activity can open
+        and locate it (#586); one under a source that is gone keeps its path
+        under that source's label. None for an operation that keeps nothing
+        (Optimize).
+        """
+        if plan.operation == "missing_links":
+            roots = _scan_roots(self._get_all_output_dirs())
+            judged = plan.payload["roots"]
+            rows = []
+            for item in db.kept_links(judged):
+                path, label, index = _scan_location(item["image_path"], roots)
+                if index is None and item["owner"] is not None:
+                    # A source that is gone is not listed: name the file
+                    # under it, with no listing index to open it by.
+                    source = judged[item["owner"]]
+                    where = [(os.path.abspath(source), self._public_path(source), None)]
+                    path, label, index = _scan_location(item["image_path"], where)
+                rows.append(
+                    {
+                        "path": path,
+                        "root": label,
+                        "root_index": index,
+                        "detail": f"Kept: {item['reason']}",
+                    }
+                )
+            return rows
+        if "kept" not in result:
+            return None
+        return [
+            {
+                "path": "",
+                "detail": f"Prompt {item['prompt_id']}: kept, "
+                + "; ".join(item["reasons"]),
+            }
+            for item in result["kept"]
+        ]
+
     def _maintenance_rows(self, operation, described):
         """Outcome rows for the changed records (root-relative paths only)."""
         if operation == "missing_links":

py/api/workflow_storage.py

@@ -59,6 +59,9 @@ REWRITE_FAILED = (
 )
 RECLAIM_FAILED = "Reclaim space failed; the database is unchanged."
 ROW_KEPT = "could not be rewritten; it keeps its inline text and reads as before"
+# A protected value's row (#586), in the review's "not text; kept as is".
+VALUE_KEPT = "stored value not text; kept as is"
+KEPT_COLUMNS = {"workflow_data": "workflow", "prompt_metadata": "prompt graph"}
 
 
 NOTHING_LEFT = "Every image record already stores its workflow once"
@@ -177,7 +180,19 @@ class WorkflowStorageRoutesMixin:
             protected += batch["protected"]
             errors += len(batch["errors"])
             job.record_counted("changed", batch["rows"])
-            job.record_counted("protected", batch["protected"])
+            job.record_counted(
+                "protected",
+                batch["protected"],
+                [
+                    {
+                        "path": "",
+                        "detail": f"Image record {image_id}: "
+                        f"{KEPT_COLUMNS.get(column, column)}, {VALUE_KEPT}",
+                    }
+                    for image_id, column in batch["kept"]
+                ],
+                complete=True,
+            )
             for image_id in batch["errors"]:
                 job.record(
                     "errors",

tests/test_api_duplicate_files.py

@@ -452,6 +452,40 @@ class TestReview(DuplicateFilesTestCase):
         )
 
 
+class TestReviewProtectedRows(DuplicateFilesTestCase):
+    async def test_cleanup_lists_each_copy_the_review_kept(self):
+        if not TestReview._can_symlink(self.tmpdir):  # pragma: no cover
+            self.skipTest("symlinks are not available")
+        self.trio()
+        job, group = await self.found_group()
+        # b.png becomes a link after the find: the review protects it (#586).
+        (self.output_dir / "b.png").unlink()
+        os.symlink(str(self.output_dir / "a.png"), str(self.output_dir / "b.png"))
+        plan = await self.review(job.id, [self.keep_of(group, "a.png")])
+        self.assertEqual(plan["protected"]["count"], 1)
+        body = await self.apply(plan["token"])
+        cleanup = self.api._jobs.get(body["job"]["id"])
+        await asyncio.wait_for(asyncio.shield(cleanup.task), DISK_BOUND_SECONDS)
+        self.assertEqual(cleanup.state, "finished", cleanup.log)
+        rows = {row["path"]: row for row in cleanup.outcomes["kept"]}
+        self.assertEqual(
+            rows["b.png"],
+            {
+                "path": "b.png",
+                "root": rows["a.png"]["root"],
+                "root_index": 0,
+                "detail": "Kept: a link, not a regular file",
+            },
+        )
+        self.assertEqual(rows["a.png"]["detail"], "Kept copy")
+        self.assertEqual(cleanup.counters, {"removed": 1, "kept": 2, "errors": 0})
+        report = cleanup.report()["outcomes"]["kept"]
+        self.assertEqual(
+            sorted(row["path"] for row in report["rows"]), ["a.png", "b.png"]
+        )
+        self.assertFalse(report["truncated"])
+
+
 class TestCleanupJob(DuplicateFilesTestCase):
     async def test_cleanup_is_a_records_job_with_a_safety_backup(self):
         self.trio()

tests/test_api_log_cleanup.py

@@ -266,6 +266,30 @@ class TestApply(LogCleanupTestCase):
         report = json.dumps(job.report())
         self.assertNotIn(self.tmpdir, report)
 
+    async def test_apply_lists_each_kept_file(self):
+        linked = self.seed()
+        plan = await self.preview()
+        _summary, job = await self.run_apply(plan)
+        expected = [
+            ("prompt_manager.log", "Kept: the active log"),
+            ("prompt_manager.log.9", "Kept: another file"),
+            ("prompt_manager.log.bak", "Kept: another file"),
+        ]
+        if linked:
+            expected.append(("prompt_manager.log.4", "Kept: a link"))
+        rows = job.outcomes["protected"]
+        self.assertEqual(
+            sorted((row["path"], row["detail"]) for row in rows), sorted(expected)
+        )
+        self.assertEqual(
+            {(row["root"], row["root_index"]) for row in rows}, {("", None)}
+        )
+        report = job.report()["outcomes"]["protected"]
+        self.assertEqual(
+            (report["count"], len(report["rows"]), report["truncated"]),
+            (plan["protected"]["count"], len(expected), False),
+        )
+
     async def test_apply_rows_name_each_file(self):
         self.seed()
         plan = await self.preview()

tests/test_api_maintenance_protected_rows.py

@@ -0,0 +1,140 @@
+"""Protected maintenance items are listed in Activity, not only counted (#586).
+
+Each record a maintenance job keeps gets one ``protected`` row with its
+reason, in the wording of that operation's review, and the job's
+downloadable report lists every one of them (never cut at 200 rows). A kept
+image link names its file, so Activity can still open and locate it.
+"""
+
+import os
+import shutil
+import sys
+from pathlib import Path
+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.jobs import MAX_OUTCOME_ROWS  # noqa: E402
+from py.config import GalleryConfig, PromptManagerConfig  # noqa: E402
+
+try:
+    from tests.test_api_maintenance import ApplyTestCase  # noqa: E402
+except ImportError:  # discovered with tests/ as the top-level directory
+    from test_api_maintenance import ApplyTestCase  # noqa: E402
+
+
+def _record(detail):
+    return {"path": "", "root": "", "root_index": None, "detail": detail}
+
+
+class TestProtectedRows(ApplyTestCase):
+    def duplicate(self, text, created_at="2999-01-01"):
+        with self.api.db.model.get_connection() as conn:
+            return conn.execute(
+                "INSERT INTO prompts(text, created_at) VALUES(?, ?)",
+                (text.upper(), created_at),
+            ).lastrowid
+
+    def assert_complete(self, job, count):
+        self.assertEqual(job.counters["protected"], count)
+        report = job.report()["outcomes"]["protected"]
+        self.assertEqual(
+            (report["count"], len(report["rows"]), report["truncated"]),
+            (count, count, False),
+        )
+
+    async def test_duplicate_prompts_lists_each_kept_copy(self):
+        paired = self.save("A dog")
+        self.api.db.save_prompt_pair("A dog", "")
+        paired_copy = self.duplicate("A dog")
+        titled = self.save("A bird")
+        titled_copy = self.duplicate("A bird")
+        with self.api.db.model.get_connection() as conn:
+            conn.execute(
+                "INSERT INTO prompt_library_state(prompt_id,title) VALUES(?,?)",
+                (titled_copy, "Bird recipe"),
+            )
+        plan, _summary, job = await self.run_apply("duplicate_prompts")
+        self.assertEqual(job.state, "finished", job.log)
+        self.assertEqual(
+            sorted(row["detail"] for row in job.outcomes["protected"]),
+            sorted(
+                [
+                    f"Prompt {paired}: kept, belongs to a saved pair",
+                    f"Prompt {paired_copy}: kept, belongs to a saved pair",
+                    f"Prompt {titled}: kept, a copy has its own Library title",
+                    f"Prompt {titled_copy}: kept, a copy has its own Library title",
+                ]
+            ),
+        )
+        self.assertEqual(plan["protected"]["count"], 4)
+        self.assert_complete(job, 4)
+
+    async def test_unused_components_lists_each_kept_prompt_with_its_reasons(self):
+        self.save("Orphan")
+        tagged = self.save("Kept", tags=["__protected__"])
+        self.api.db.save_prompt_pair("Paired", "")
+        paired = self.api.db.search_prompts(text="Paired")[0]["id"]
+        with self.api.db.model.get_connection() as conn:
+            conn.execute(
+                "INSERT INTO prompt_tags(prompt_id, tag_id) "
+                "SELECT ?, id FROM tags WHERE name='__protected__'",
+                (paired,),
+            )
+        plan, _summary, job = await self.run_apply("unused_components")
+        self.assertEqual(job.state, "finished", job.log)
+        self.assertEqual(
+            sorted(job.outcomes["protected"], key=lambda row: row["detail"]),
+            sorted(
+                [
+                    _record(f"Prompt {tagged}: kept, tagged __protected__"),
+                    _record(
+                        f"Prompt {paired}: kept, in a saved pair; "
+                        "tagged __protected__"
+                    ),
+                ],
+                key=lambda row: row["detail"],
+            ),
+        )
+        self.assertEqual(plan["protected"]["count"], 2)
+        self.assert_complete(job, 2)
+
+    async def test_report_lists_every_kept_prompt(self):
+        PromptManagerConfig.SAFETY_BACKUP = False
+        count = MAX_OUTCOME_ROWS + 5
+        for n in range(count):
+            self.save(f"Kept {n:03d}", tags=["__protected__"])
+        _plan, _summary, job = await self.run_apply("unused_components")
+        self.assertEqual(len(job.outcomes["protected"]), MAX_OUTCOME_ROWS)
+        self.assert_complete(job, count)
+
+    async def test_missing_links_lists_each_kept_link_with_its_file(self):
+        unmounted = Path(self.tmpdir) / "unmounted"
+        unmounted.mkdir()
+        GalleryConfig.MONITORING_DIRECTORIES = [str(self.output_dir), str(unmounted)]
+        prompt = self.save("Links")
+        self.image(prompt, "gone.png", exists=False)
+        self.image(prompt, "sub/away.png", folder=unmounted, exists=False)
+        self.image(prompt, "x.png", folder=Path(self.tmpdir) / "elsewhere")
+        shutil.rmtree(unmounted)
+        plan, _summary, job = await self.run_apply("missing_links")
+        self.assertEqual(job.state, "finished", job.log)
+        self.assertEqual(plan["protected"]["count"], 2)
+        rows = sorted(job.outcomes["protected"], key=lambda row: row["path"])
+        self.assertEqual(
+            [(row["path"], row["detail"]) for row in rows],
+            [
+                ("sub/away.png", "Kept: link under an unavailable source"),
+                ("x.png", "Kept: link outside the image sources"),
+            ],
+        )
+        # The gone source keeps the path under its label; outside every
+        # source the row names the file only.
+        self.assertEqual((rows[0]["root"], rows[0]["root_index"]), ("unmounted", None))
+        self.assertEqual((rows[1]["root"], rows[1]["root_index"]), ("", None))
+        self.assertNotIn(self.tmpdir, str(job.report()))
+        self.assert_complete(job, 2)

tests/test_api_workflow_storage.py

@@ -363,11 +363,11 @@ class TestWorkflowStorageOutcomes(WorkflowStorageTestCase):
         prompt_id = self.old_library(count=4)
         conn = self.api.db.model.get_connection()
         with conn:
-            conn.execute(
+            image_id = conn.execute(
                 "INSERT INTO generated_images (prompt_id, image_path, filename, "
                 "file_path, workflow_data) VALUES (?, '/x/b', 'b', '/x/b', x'00')",
                 (prompt_id,),
-            )
+            ).lastrowid
             store_text(conn, "a document no record references")
         _plan, _summary, job = await self.start()
         await self.finish(job)
@@ -377,6 +377,13 @@ class TestWorkflowStorageOutcomes(WorkflowStorageTestCase):
             "Rewrote 4 image records; 1 stored value not text, kept as is", log
         )
         self.assertIn("Removed 1 unused workflow document", log)
+        # #586: the kept value has its row, in the review's wording.
+        self.assertEqual(
+            [row["detail"] for row in job.outcomes["protected"]],
+            [f"Image record {image_id}: workflow, stored value not text; kept as is"],
+        )
+        report = job.report()["outcomes"]["protected"]
+        self.assertEqual((report["count"], report["truncated"]), (1, False))
 
     async def reclaim_with(self, **patches):
         self.old_library(count=4)

tests/test_jobs.py

@@ -506,6 +506,20 @@ class TestJobOutcomesAndLog(unittest.TestCase):
         self.assertFalse(report["changed"]["truncated"])
         self.assertEqual(report["errors"], {"count": 0, "rows": [], "truncated": False})
 
+    def test_record_into_a_complete_group_keeps_the_row_for_the_report(self):
+        # #586: a group kept whole stays whole when later rows are recorded.
+        job = Job("duplicate_cleanup", "Remove", groups=("kept",))
+        rows = [
+            {"path": f"f{i}.png", "detail": "Kept"} for i in range(MAX_OUTCOME_ROWS)
+        ]
+        job.record_counted("kept", len(rows), rows, complete=True)
+        job.record("kept", {"path": "late.png", "detail": "Kept copy"}, "= late.png")
+        self.assertEqual(len(job.outcomes["kept"]), MAX_OUTCOME_ROWS)
+        report = job.report()["outcomes"]["kept"]
+        self.assertEqual(report["count"], MAX_OUTCOME_ROWS + 1)
+        self.assertEqual(report["rows"][-1]["path"], "late.png")
+        self.assertFalse(report["truncated"])
+
     def test_thumbnail_job_groups(self):
         self.assertEqual(
             jobs_module.THUMBNAIL_GROUP_NAMES, ("generated", "skipped", "errors")