feat(library): archive paired and delete unpaired selected prompts on the server (#443)

A prompt_archive table marks Library parents hidden from the shelf; older
builds ignore it. New routes /prompt_manager/library/delete/preview,
/library/delete and /library/restore classify a selection (delete,
archive, protected, already archived, not found), execute exactly what was
reviewed under a records hold, take a verified safety backup (disk checked)
before any delete, and restore archived prompts. The shelf, search and the
shelf-facing counts honor the archive (?shelf=active|archived); archiving
repairs a stale prompt hash so scans find the archived row.

database/library.py

@@ -6,6 +6,7 @@ from collections.abc import Iterator
 from contextlib import contextmanager
 
 from .library_tokens import library_edit_token
+from .prompt_archive import ARCHIVED, SHELVED
 
 _ELIGIBLE = """(EXISTS(SELECT 1 FROM prompt_pairs ep WHERE ep.positive_prompt_id=p.id)
  OR EXISTS(SELECT 1 FROM generated_images eg WHERE eg.prompt_id=p.id)
@@ -105,6 +106,12 @@ ARCHIVE_SCOPES = {
 }
 
 
+# Shelves (#443): the parents on the shelf, or the archived ones.
+PARENT_SCOPES = {"active": SHELVED, "archived": ARCHIVED}
+# A positive pair owner that is on the shelf, as a condition on alias pp.
+SHELVED_PAIR = SHELVED.replace("p.id", "pp.positive_prompt_id")
+
+
 def _escape(value: str) -> str:
     return value.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
 
@@ -216,6 +223,8 @@ class LibraryOperations:
         rows = conn.execute(
             "SELECT p.id,p.text,ls.title,p.category,p.rating,p.notes,p.created_at,"
             "p.run_count,p.last_used_at,"
+            + ARCHIVED
+            + " AS archived,"
             + (f"{sort_expression} AS _sort," if sort_expression else "")
             + "(SELECT COUNT(*) FROM prompt_pairs v WHERE v.positive_prompt_id=p.id "
             "AND v.archived=0) AS variant_count,"
@@ -239,6 +248,7 @@ class LibraryOperations:
             # Usage is shown, never compared: the token hashes owner fields only.
             row["run_count"] = row["run_count"] if type(row["run_count"]) is int else 0
             row["last_used_at"] = utc_iso(row["last_used_at"])
+            row["archived"] = bool(row["archived"])
             row["edit_token"] = library_edit_token(epoch, row)
         return result
 
@@ -254,10 +264,13 @@ class LibraryOperations:
         sort: str = "newest",
         limit: int = 40,
         after: dict | None = None,
+        shelf: str = "active",
     ) -> dict:
         _limit(limit)
         if sort not in _ORDERS:
             raise ValueError("Unknown library sort order")
+        if not isinstance(shelf, str) or shelf not in PARENT_SCOPES:
+            raise ValueError("shelf must be active or archived")
         where, params = _filter(text, category, [] if tags is None else tags)
         if after is not None:
             if (
@@ -270,15 +283,18 @@ class LibraryOperations:
             ceiling = after["ceiling"]
         where += " AND p.id<=?"
         params.append(ceiling)
-        # One SELECT counts the filtered parents and their active variants.
-        total, variant_total = conn.execute(
-            "SELECT COUNT(*),COALESCE(SUM((SELECT COUNT(*) FROM prompt_pairs v "
-            "WHERE v.positive_prompt_id=p.id AND v.archived=0)),0) FROM "
-            + _FROM
-            + " WHERE "
-            + where,
+        # One SELECT counts the filtered parents of this scope, their active
+        # variants, and the filtered archived parents (Show archived, #443).
+        scope = PARENT_SCOPES[shelf]
+        total, variant_total, archived_total = conn.execute(
+            f"SELECT COALESCE(SUM({scope}),0),"
+            f"COALESCE(SUM(CASE WHEN {scope} THEN (SELECT COUNT(*) "
+            "FROM prompt_pairs v WHERE v.positive_prompt_id=p.id "
+            "AND v.archived=0) ELSE 0 END),0),"
+            f"COALESCE(SUM({ARCHIVED}),0) FROM " + _FROM + " WHERE " + where,
             params,
         ).fetchone()
+        where += f" AND {scope}"
         expression, direction = _ORDERS[sort]
         if after is not None:
             operator = ">" if direction == "ASC" else "<"
@@ -311,6 +327,7 @@ class LibraryOperations:
             "items": rows,
             "total": total,
             "variant_total": variant_total,
+            "archived_total": archived_total,
             "has_more": has_more,
             "next_key": next_key,
         }
@@ -324,6 +341,7 @@ class LibraryOperations:
         sort: str = "newest",
         limit: int = 40,
         after: dict | None = None,
+        shelf: str = "active",
     ) -> dict:
         with self.library_snapshot(include_ceiling=True) as (conn, epoch, ceiling):
             return self._list_library(
@@ -336,6 +354,7 @@ class LibraryOperations:
                 sort=sort,
                 limit=limit,
                 after=after,
+                shelf=shelf,
             )
 
     def _is_library_positive(self, conn: sqlite3.Connection, positive_id: int) -> bool:
@@ -576,7 +595,9 @@ class LibraryOperations:
             owners = {}
             order = []
             for row in conn.execute(
-                "SELECT p.id,p.text,ls.title,p.category,p.rating,p.notes FROM "
+                "SELECT p.id,p.text,ls.title,p.category,p.rating,p.notes,"
+                + ARCHIVED
+                + " AS archived FROM "
                 + _FROM
                 + " WHERE "
                 + _ELIGIBLE
@@ -589,6 +610,7 @@ class LibraryOperations:
                     "tags": [],
                     "rating": row["rating"],
                     "notes": row["notes"],
+                    "archived": bool(row["archived"]),
                     "variants": [],
                 }
                 order.append(row["id"])
@@ -633,8 +655,9 @@ class LibraryOperations:
         """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
+        Library's ``total`` and ``variant_total``: eligible parents on the
+        shelf (not archived, #443) and their active pairs (every pair's
+        positive is eligible). ``images`` counts indexed
         ``generated_images`` rows, not files on disk. An image has no rating
         of its own (it shows its linked prompt's), so ``rated_images`` counts
         the indexed images whose linked prompt is rated. ``review_pending`` is
@@ -642,8 +665,14 @@ class LibraryOperations:
         """
         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 (SELECT COUNT(*) FROM prompts p WHERE "
+                + _ELIGIBLE
+                + " AND "
+                + SHELVED
+                + "),"
+                "(SELECT COUNT(*) FROM prompt_pairs pp WHERE pp.archived=0 AND "
+                + SHELVED_PAIR
+                + "),"
                 "(SELECT COUNT(*) FROM generated_images),"
                 "(SELECT COUNT(*) FROM generated_images gi JOIN prompts r "
                 "ON r.id=gi.prompt_id WHERE r.rating IS NOT NULL),"

database/library_archive.py

@@ -0,0 +1,289 @@
+"""Delete selected Library prompts; archive the paired ones (#443, D152).
+
+A selected Library parent is classified, in this order, as ``not_found``
+(missing, or not a Library parent), ``protected`` (tagged ``__protected__``;
+never changed), ``already_archived``, ``archive`` (the positive or negative of
+any saved pair, archived pair or not: archiving hides it from the shelf and
+removes nothing) or ``delete`` (in no pair: deleted exactly as the legacy
+delete does, through ``delete_prompt_rows``). Every command checks the
+storage epoch first and runs in one ``BEGIN IMMEDIATE`` transaction.
+
+The write executes what the user reviewed: it is given the reviewed archive
+and delete lists, and deletes only reviewed deletes the route backed up
+(``candidates``) whose fingerprint is unchanged since that backup.
+"""
+
+import sqlite3
+from typing import Iterable
+
+from .library import _ELIGIBLE
+from .library_commands import _validate_bulk_ids, _validate_epoch
+from .models import utc_now_iso
+from .prompt_archive import ARCHIVED as _ARCHIVED
+from .prompt_archive import TABLE as ARCHIVE
+from .workflow_documents import documents_of_rows, release_documents
+
+try:
+    from ..utils.hashing import generate_prompt_hash
+except ImportError:
+    from utils.hashing import generate_prompt_hash
+
+# Ids bound per statement: the pair check binds each id twice, which stays
+# under SQLite's historical 999-variable limit (maintenance.py ID_BATCH).
+ID_BATCH = 400
+_PROTECTED = (
+    "EXISTS(SELECT 1 FROM prompt_tags pt JOIN tags t ON t.id=pt.tag_id "
+    "WHERE pt.prompt_id=p.id AND t.name='__protected__')"
+)
+_PAIRED = (
+    "EXISTS(SELECT 1 FROM prompt_pairs pp "
+    "WHERE pp.positive_prompt_id=p.id OR pp.negative_prompt_id=p.id)"
+)
+# Classes in the order a prompt is tested; the first that holds wins.
+CLASSES = ("not_found", "protected", "already_archived", "archive", "delete")
+
+
+def _batches(ids: list) -> Iterable[list]:
+    for start in range(0, len(ids), ID_BATCH):
+        yield ids[start : start + ID_BATCH]
+
+
+def delete_prompt_rows(conn: sqlite3.Connection, prompt_ids: list[int]) -> int:
+    """Delete prompts and their image link rows; return the prompts deleted.
+
+    The legacy delete's statements (``delete_prompt``, ``bulk_delete_prompts``)
+    in the caller's transaction: image link rows first (``image_prompt_pairs``
+    cascades, review decisions keep a NULL image), then the prompts (tags and
+    Library state cascade), then the workflow documents no row references.
+    Image files are never touched. The caller checks pairs first.
+    """
+    documents = documents_of_rows(conn, "prompt_id", prompt_ids)
+    deleted = 0
+    for chunk in _batches(list(prompt_ids)):
+        marks = ",".join("?" for _ in chunk)
+        conn.execute(
+            f"DELETE FROM generated_images WHERE prompt_id IN ({marks})",  # nosec
+            chunk,
+        )
+        deleted += conn.execute(
+            f"DELETE FROM prompts WHERE id IN ({marks})", chunk  # nosec B608
+        ).rowcount
+    release_documents(conn, documents)
+    return deleted
+
+
+class LibraryArchiveOperations:
+    """PromptDatabase mixin: classify, archive or delete, and restore parents."""
+
+    def _classify_library_delete(
+        self, conn: sqlite3.Connection, prompt_ids: list[int]
+    ) -> dict:
+        found = {}
+        for chunk in _batches(prompt_ids):
+            marks = ",".join("?" for _ in chunk)
+            for row in conn.execute(
+                f"SELECT p.id,{_PROTECTED},{_ARCHIVED},{_PAIRED} "  # nosec B608
+                f"FROM prompts p WHERE p.id IN ({marks}) AND {_ELIGIBLE}",
+                chunk,
+            ):
+                protected, archived, paired = row[1], row[2], row[3]
+                if protected:
+                    found[row[0]] = "protected"
+                elif archived:
+                    found[row[0]] = "already_archived"
+                else:
+                    found[row[0]] = "archive" if paired else "delete"
+        classes = {name: [] for name in CLASSES}
+        for pid in prompt_ids:
+            classes[found.get(pid, "not_found")].append(pid)
+        return classes
+
+    @staticmethod
+    def _image_links(conn: sqlite3.Connection, prompt_ids: list[int]) -> int:
+        total = 0
+        for chunk in _batches(prompt_ids):
+            marks = ",".join("?" for _ in chunk)
+            total += conn.execute(
+                "SELECT COUNT(*) FROM generated_images "  # nosec B608
+                f"WHERE prompt_id IN ({marks})",
+                chunk,
+            ).fetchone()[0]
+        return total
+
+    @staticmethod
+    def _fingerprints(conn: sqlite3.Connection, prompt_ids: list[int]) -> dict:
+        """What a safety backup holds of each prompt a delete would remove.
+
+        The prompt row (usage included), its tag ids, its Library title and
+        its image link rows (count and newest id), as a JSON-safe list.
+        """
+        prints = {}
+        for chunk in _batches(prompt_ids):
+            marks = ",".join("?" for _ in chunk)
+            for row in conn.execute(
+                "SELECT p.id,p.text,p.hash,p.category,p.rating,p.notes,"  # nosec
+                "p.updated_at,p.run_count,p.last_used_at,"
+                "(SELECT group_concat(tag_id) FROM (SELECT tag_id FROM prompt_tags "
+                "WHERE prompt_id=p.id ORDER BY tag_id)),"
+                "(SELECT title FROM prompt_library_state WHERE prompt_id=p.id),"
+                "(SELECT COUNT(*) FROM generated_images WHERE prompt_id=p.id),"
+                "(SELECT MAX(id) FROM generated_images WHERE prompt_id=p.id) "
+                f"FROM prompts p WHERE p.id IN ({marks})",
+                chunk,
+            ):
+                prints[row[0]] = [
+                    value if isinstance(value, (int, str, type(None))) else repr(value)
+                    for value in tuple(row)[1:]
+                ]
+        return prints
+
+    @staticmethod
+    def _checked_ids(prompt_ids: list[int], expected_epoch: str) -> list[int]:
+        _validate_epoch(expected_epoch)
+        _validate_bulk_ids(prompt_ids)
+        return list(dict.fromkeys(prompt_ids))
+
+    def classify_library_delete(
+        self, prompt_ids: list[int], *, expected_epoch: str
+    ) -> dict:
+        """What a delete of ``prompt_ids`` would do now (reads only).
+
+        Returns the id lists per class, in request order, ``image_links``
+        (the image link rows the ``delete`` class would remove) and
+        ``fingerprints`` of the ``delete`` class (``_fingerprints``).
+        """
+        prompt_ids = self._checked_ids(prompt_ids, expected_epoch)
+        with self.library_snapshot() as (conn, _epoch, _ceiling):
+            epoch = self._library_epoch(conn, expected_epoch)
+            classes = self._classify_library_delete(conn, prompt_ids)
+            return {
+                "library_epoch": epoch,
+                **classes,
+                "image_links": self._image_links(conn, classes["delete"]),
+                "fingerprints": self._fingerprints(conn, classes["delete"]),
+            }
+
+    @staticmethod
+    def _repair_hash(conn: sqlite3.Connection, prompt_id: int) -> None:
+        """Give an archived prompt its canonical hash, so ingestion finds it.
+
+        Scans, the monitor and node saves resolve prompts by hash; a stale or
+        missing one would make them insert a visible copy (Q3). When another
+        row already holds the canonical hash, lookups find that row and no new
+        copy is made, so nothing is written.
+        """
+        text, stored = conn.execute(
+            "SELECT text,hash FROM prompts WHERE id=?", (prompt_id,)
+        ).fetchone()
+        canonical = generate_prompt_hash(text) if isinstance(text, str) else None
+        if canonical is None or stored == canonical:
+            return
+        if conn.execute(
+            "SELECT 1 FROM prompts WHERE hash=? AND id!=?", (canonical, prompt_id)
+        ).fetchone():
+            return
+        conn.execute("UPDATE prompts SET hash=? WHERE id=?", (canonical, prompt_id))
+
+    def archive_or_delete_library_prompts(
+        self,
+        archive_ids: list[int],
+        delete_ids: list[int],
+        *,
+        expected_epoch: str,
+        candidates: dict | None,
+    ) -> dict:
+        """Archive and delete what the user reviewed, all or nothing.
+
+        ``archive_ids`` and ``delete_ids`` are the reviewed classes.
+        ``candidates`` maps each reviewed delete the route backed up to its
+        fingerprint then (None: every reviewed delete, unchecked; tests and
+        callers without a backup). A reviewed delete is deleted only when it
+        still classifies ``delete`` and its fingerprint is unchanged; one that
+        is now paired, and a reviewed archive now unpaired, are archived (the
+        restorable direction); anything else is kept and counted.
+        """
+        _validate_epoch(expected_epoch)
+        for ids in (archive_ids, delete_ids):
+            if not isinstance(ids, list) or any(
+                type(pid) is not int or not 1 <= pid <= 2**63 - 1 for pid in ids
+            ):
+                raise ValueError("prompt ids must be lists of positive integers")
+        reviewed_delete = set(delete_ids)
+        if not archive_ids and not delete_ids:
+            raise ValueError("Nothing to delete or archive")
+        if reviewed_delete & set(archive_ids):
+            raise ValueError("A prompt cannot be both archived and deleted")
+        prompt_ids = list(dict.fromkeys([*archive_ids, *delete_ids]))
+        with self._pair_transaction() as conn:
+            epoch = self._library_epoch(conn, expected_epoch)
+            classes = self._classify_library_delete(conn, prompt_ids)
+            now_prints = self._fingerprints(conn, classes["delete"])
+            deletes, archives, changed = [], list(classes["archive"]), 0
+            for pid in classes["delete"]:
+                if pid not in reviewed_delete:
+                    archives.append(pid)  # reviewed as archive, now unpaired
+                elif candidates is None or candidates.get(pid) == now_prints[pid]:
+                    deletes.append(pid)
+                else:
+                    changed += 1
+            changed += len(classes["protected"])
+            image_links = self._image_links(conn, deletes)
+            now = utc_now_iso()
+            for pid in archives:
+                self._repair_hash(conn, pid)
+            conn.executemany(
+                f"INSERT INTO {ARCHIVE}(prompt_id, archived_at) VALUES (?, ?)",
+                [(pid, now) for pid in archives],
+            )
+            # A changed row is the newest edit the restore fingerprint reads.
+            conn.executemany(
+                "UPDATE prompts SET updated_at=? WHERE id=?",
+                [(now, pid) for pid in archives],
+            )
+            deleted = delete_prompt_rows(conn, deletes)
+            return {
+                "library_epoch": epoch,
+                "deleted": deleted,
+                "archived": len(archives),
+                "already_archived": len(classes["already_archived"]),
+                "not_found": len(classes["not_found"]),
+                "changed": changed,
+                "image_links": image_links,
+            }
+
+    def restore_library_prompts(
+        self, prompt_ids: list[int], *, expected_epoch: str
+    ) -> dict:
+        """Put archived prompts back on the shelf, all or nothing.
+
+        Every requested id with an archive row is restored, whatever its
+        protection or Library parent state: restoring only un-hides it.
+        """
+        prompt_ids = self._checked_ids(prompt_ids, expected_epoch)
+        with self._pair_transaction() as conn:
+            epoch = self._library_epoch(conn, expected_epoch)
+            restored, existing = [], set()
+            for chunk in _batches(prompt_ids):
+                marks = ",".join("?" for _ in chunk)
+                for pid, archived in conn.execute(
+                    f"SELECT p.id,{_ARCHIVED} FROM prompts p "  # nosec B608
+                    f"WHERE p.id IN ({marks})",
+                    chunk,
+                ):
+                    existing.add(pid)
+                    if archived:
+                        restored.append(pid)
+            now = utc_now_iso()
+            conn.executemany(
+                f"DELETE FROM {ARCHIVE} WHERE prompt_id=?", [(pid,) for pid in restored]
+            )
+            conn.executemany(
+                "UPDATE prompts SET updated_at=? WHERE id=?",
+                [(now, pid) for pid in restored],
+            )
+            return {
+                "library_epoch": epoch,
+                "restored": len(restored),
+                "unchanged": len(existing) - len(restored),
+                "not_found": len(prompt_ids) - len(existing),
+            }

database/maintenance.py

@@ -80,6 +80,12 @@ PROTECTIONS = (
         "with Library state (a title or an explicit parent)",
         "with Library state (a title or an explicit parent)",
     ),
+    (
+        # #443: an archived prompt is restorable, never pruned.
+        "EXISTS (SELECT 1 FROM prompt_archive pa WHERE pa.prompt_id = p.id)",
+        "archived in the Library",
+        "archived in the Library",
+    ),
     (
         "p.id IN (SELECT prompt_id FROM live_hint_prompts)",
         "used by recorded runs",
@@ -88,7 +94,8 @@ PROTECTIONS = (
 )
 
 # Exempt references: images, pairs, the __protected__ tag, Library parent
-# state, and live recorded-run hints (computed once, see LIVE_HINT_CTES).
+# state, the Library archive, and live recorded-run hints (computed once,
+# see LIVE_HINT_CTES).
 ORPHANED_PROMPTS_SQL = (
     "WITH "
     + LIVE_HINT_CTES

database/models.py

@@ -39,6 +39,7 @@ from .review_schema import (
 )
 from .saved_views import migrate_saved_views, saved_views_schema_error
 from .image_categories import image_categories_schema_error, migrate_image_categories
+from .prompt_archive import migrate_prompt_archive, prompt_archive_schema_error
 from .workflow_documents import (
     DOCUMENT_ID_COLUMNS,
     migrate_workflow_documents,
@@ -358,6 +359,8 @@ class PromptModel:
         migrate_autotag_state(conn)
         # Image categories (#332): independent of every table above.
         migrate_image_categories(conn)
+        # The Library parent archive (#443): independent of every table above.
+        migrate_prompt_archive(conn)
         # Last: the older generated_images rebuilds run only before file_path.
         migrate_workflow_documents(conn)
 
@@ -1234,6 +1237,11 @@ class PromptModel:
             )
             if categories_error:
                 return False, categories_error
+            archive_error = prompt_archive_schema_error(
+                conn, allow_missing=not require_library_state
+            )
+            if archive_error:
+                return False, archive_error
             views_error = saved_views_schema_error(
                 conn, allow_missing=not require_library_state
             )

database/operations.py

@@ -24,6 +24,7 @@ from .review import ReviewOperations
 from .review_decisions import ReviewDecisionOperations
 from .saved_views import SavedViewOperations
 from .library_commands import LibraryCommandOperations
+from .library_archive import LibraryArchiveOperations, delete_prompt_rows
 from .vocabulary import VocabularyOperations
 from .execution_history import ExecutionHistoryOperations
 from .execution_contexts import ExecutionContextOperations
@@ -448,6 +449,7 @@ class PromptDatabase(
     PairOperations,
     LibraryOperations,
     LibraryCommandOperations,
+    LibraryArchiveOperations,
     VocabularyOperations,
     ExecutionHistoryOperations,
     ExecutionContextOperations,
@@ -2344,16 +2346,10 @@ class PromptDatabase(
         Returns:
             Number of prompts actually deleted.
         """
-        count = 0
         with self._pair_transaction() as conn:
             self._ensure_prompts_unpaired(conn, prompt_ids)
-            documents = documents_of_rows(conn, "prompt_id", prompt_ids)
-            for pid in prompt_ids:
-                conn.execute("DELETE FROM generated_images WHERE prompt_id = ?", (pid,))
-                cursor = conn.execute("DELETE FROM prompts WHERE id = ?", (pid,))
-                if cursor.rowcount > 0:
-                    count += 1
-            release_documents(conn, documents)
+            # The workspace Library's delete runs the same statements (#443).
+            count = delete_prompt_rows(conn, list(dict.fromkeys(prompt_ids)))
             conn.commit()
         return count
 

database/prompt_archive.py

@@ -0,0 +1,107 @@
+"""Library parent archive (#443, D152): prompts hidden from the shelf, restorable.
+
+A row means the Library parent is archived; no row means it is on the shelf.
+Nothing else about the prompt changes: its text, metadata, tags, run count,
+image links, pairs and recorded runs stay. Restore deletes the row. Deleting
+the prompt (any path) drops the row by cascade.
+
+A table of its own, not a column: ``prompt_library_state`` is validated by its
+exact definition, so a new column there would make older builds refuse the
+upgraded database, while older builds ignore an unknown table (they show the
+archived prompts again and keep the rows). Optional in old backups, required
+once a database is current.
+"""
+
+import re
+import sqlite3
+
+TABLE = "prompt_archive"
+MAX_STAMP_BYTES = 64
+# What ``utc_now_iso`` writes starts with a date; anything else was not written here.
+_STAMP_GLOB = "[0-9][0-9][0-9][0-9]-[0-9][0-9]-[0-9][0-9]T*"
+_DDL = (
+    f"CREATE TABLE {TABLE} (prompt_id INTEGER PRIMARY KEY, "
+    "archived_at TEXT NOT NULL, "
+    "FOREIGN KEY(prompt_id) REFERENCES prompts(id) ON DELETE CASCADE)"
+)
+_COLUMNS = (("prompt_id", "INTEGER", 0, 1), ("archived_at", "TEXT", 1, 0))
+_FOREIGN_KEYS = {("prompt_id", "prompts", "id", "CASCADE")}
+# A predicate on ``p`` (the prompts alias): the parent is not archived.
+SHELVED = f"NOT EXISTS(SELECT 1 FROM {TABLE} pa WHERE pa.prompt_id=p.id)"
+ARCHIVED = f"EXISTS(SELECT 1 FROM {TABLE} pa WHERE pa.prompt_id=p.id)"
+
+
+def _compact(sql: str) -> str:
+    return re.sub(r"\s+", "", sql or "").lower()
+
+
+def _structure_error(conn: sqlite3.Connection, allow_missing: bool) -> str | None:
+    objects = conn.execute(
+        "SELECT type,sql FROM sqlite_master WHERE name=? "
+        "OR (type IN ('index','trigger') AND tbl_name=?)",
+        (TABLE, TABLE),
+    ).fetchall()
+    if not objects:
+        return None if allow_missing else "Missing prompt archive schema"
+    tables = [sql for kind, sql in objects if kind == "table"]
+    if len(tables) != 1 or len(objects) != 1:
+        return "Prompt archive name is reserved by an unsupported schema object"
+    if _compact(tables[0]) != _compact(_DDL):
+        return "Prompt archive has an invalid table definition"
+    columns = tuple(
+        (row[1], row[2].upper(), row[3], row[5])
+        for row in conn.execute(f"PRAGMA table_info({TABLE})")
+    )
+    if columns != _COLUMNS:
+        return "Prompt archive has invalid columns"
+    foreign_keys = {
+        (row[3], row[2], row[4], row[6])
+        for row in conn.execute(f"PRAGMA foreign_key_list({TABLE})")
+    }
+    if foreign_keys != _FOREIGN_KEYS:
+        return "Prompt archive has invalid foreign key definitions"
+    return None
+
+
+def prompt_archive_schema_error(
+    conn: sqlite3.Connection, *, allow_missing: bool = True
+) -> str | None:
+    """Reject an unsupported archive schema or values; reads only.
+
+    A missing table is acceptable for backups that predate it unless
+    ``allow_missing`` is false.
+    """
+    error = _structure_error(conn, allow_missing)
+    if error is not None:
+        return error
+    if not conn.execute(
+        "SELECT 1 FROM sqlite_master WHERE type='table' AND name=?", (TABLE,)
+    ).fetchone():
+        return None
+    if conn.execute(f"PRAGMA foreign_key_check({TABLE})").fetchone():
+        return "Prompt archive contains foreign key violations"
+    if conn.execute(
+        f"SELECT 1 FROM {TABLE} WHERE typeof(archived_at)!='text' "  # nosec B608
+        f"OR length(CAST(archived_at AS BLOB))>{MAX_STAMP_BYTES} "
+        f"OR archived_at NOT GLOB '{_STAMP_GLOB}' LIMIT 1"
+    ).fetchone():
+        return "Prompt archive contains invalid values"
+    return None
+
+
+def migrate_prompt_archive(conn: sqlite3.Connection) -> None:
+    """Install the table atomically; validate before and after (no backfill)."""
+    conn.execute("SAVEPOINT prompt_archive_schema")
+    try:
+        error = _structure_error(conn, True)
+        if error:
+            raise sqlite3.DatabaseError(error)
+        conn.execute(_DDL.replace("CREATE TABLE", "CREATE TABLE IF NOT EXISTS", 1))
+        error = _structure_error(conn, False)
+        if error:
+            raise sqlite3.DatabaseError(error)
+        conn.execute("RELEASE prompt_archive_schema")
+    except BaseException:
+        conn.execute("ROLLBACK TO prompt_archive_schema")
+        conn.execute("RELEASE prompt_archive_schema")
+        raise

database/restore_stage.py

@@ -149,6 +149,7 @@ def library_summary(conn: sqlite3.Connection) -> Dict[str, int]:
     collections (#331).
     """
     from .library import _ELIGIBLE
+    from .prompt_archive import SHELVED, TABLE as ARCHIVE
 
     has_state = conn.execute(
         "SELECT 1 FROM sqlite_master WHERE type='table' AND "
@@ -157,9 +158,15 @@ def library_summary(conn: sqlite3.Connection) -> Dict[str, int]:
     has_pairs = conn.execute(
         "SELECT 1 FROM sqlite_master WHERE type='table' AND name='prompt_pairs'"
     ).fetchone()
+    has_archive = conn.execute(
+        "SELECT 1 FROM sqlite_master WHERE type='table' AND name=?", (ARCHIVE,)
+    ).fetchone()
     if has_state and has_pairs:
+        # Archived parents (#443) are off the shelf, as in library_counts.
         positives = conn.execute(
-            "SELECT COUNT(*) FROM prompts p WHERE " + _ELIGIBLE
+            "SELECT COUNT(*) FROM prompts p WHERE "
+            + _ELIGIBLE
+            + (" AND " + SHELVED if has_archive else "")
         ).fetchone()[0]
     else:
         positives = _count(conn, "prompts")

database/vocabulary.py

@@ -20,7 +20,8 @@ counted, hashed into the token and renamed with the prompts and pairs.
 
 import sqlite3
 
-from .library import _ELIGIBLE, _limit
+from .library import _ELIGIBLE, SHELVED_PAIR, _limit
+from .prompt_archive import SHELVED
 from .library_commands import _TOKEN, LibraryCommandError, _validate_epoch
 from .library_tokens import canonical_token
 from .models import utc_now_iso
@@ -40,17 +41,21 @@ _IMAGE_FILES = (
 
 def _tag_rows(page: str) -> str:
     """Count rows for the registry rows selected by ``page`` (id, name)."""
-    # Eligible parent components carrying a tag in the page, plus the
-    # positive parent of every active pair carrying it (always eligible).
+    # Eligible parent components on the shelf carrying a tag in the page,
+    # plus the positive parent of every active pair carrying it (always
+    # eligible) when that parent is on the shelf: archived parents (#443)
+    # are hidden with their pairs, so Browse lists what was counted.
     return (
         "WITH page AS (" + page + "),"
         "owners(tag_id,pid) AS ("
         "SELECT pt.tag_id,p.id FROM prompts p "
         "JOIN prompt_tags pt ON pt.prompt_id=p.id "
         "WHERE pt.tag_id IN (SELECT id FROM page) AND " + _ELIGIBLE + " "
+        "AND " + SHELVED + " "
         "UNION SELECT ppt.tag_id,pp.positive_prompt_id FROM prompt_pairs pp "
         "JOIN prompt_pair_tags ppt ON ppt.pair_id=pp.id "
-        "WHERE pp.archived=0 AND ppt.tag_id IN (SELECT id FROM page)),"
+        "WHERE pp.archived=0 AND " + SHELVED_PAIR + " "
+        "AND ppt.tag_id IN (SELECT id FROM page)),"
         "images(tag_id,n) AS (" + _IMAGE_FILES + " "
         "WHERE mit.tag_id IN (SELECT id FROM page) GROUP BY mit.tag_id) "
         "SELECT page.id,page.name,COUNT(DISTINCT owners.pid) AS prompt_count,"
@@ -100,7 +105,9 @@ def _validate_tag_id(tag_id: object) -> None:
 
 
 # Category values in Tags & categories (#332 §7): the prompt-backed values
-# (nonblank values on eligible parents or on active pairs, grouped exactly, so
+# (nonblank values on eligible parents or on active pairs; ``prompt_count``
+# counts only parents on the shelf, so a value only archived parents (#443)
+# hold stays listed with 0, as a tag does; grouped exactly, so
 # whitespace and case variants stay distinct; /library/options lists the same
 # prompt-backed set with its own query) plus the values only image rows hold.
 # A prompt-backed value is ordered by its oldest owning positive; an image-only
@@ -109,15 +116,17 @@ def _validate_tag_id(tag_id: object) -> None:
 # files missing on disk. No FULL OUTER JOIN, so it runs on SQLite 3.31+.
 _IMAGE_ONLY_FIRST_PID = 2**63 - 1
 _CATEGORY_ROWS = (
-    "WITH owners(value,pid) AS ("
-    "SELECT p.category,p.id FROM prompts p WHERE p.category IS NOT NULL "
-    "AND TRIM(p.category)!='' AND " + _ELIGIBLE + " "
-    "UNION SELECT pp.category,pp.positive_prompt_id FROM prompt_pairs pp "
+    "WITH owners(value,pid,shelved) AS ("
+    "SELECT p.category,p.id," + SHELVED + " FROM prompts p "
+    "WHERE p.category IS NOT NULL AND TRIM(p.category)!='' AND " + _ELIGIBLE + " "
+    "UNION SELECT pp.category,pp.positive_prompt_id," + SHELVED_PAIR + " "
+    "FROM prompt_pairs pp "
     "WHERE pp.archived=0 AND pp.category IS NOT NULL AND TRIM(pp.category)!=''),"
     "image_values(value,image_count) AS ("
     "SELECT category,COUNT(*) FROM image_categories GROUP BY category),"
     "grouped(value,first_pid,prompt_count) AS ("
-    "SELECT value,MIN(pid),COUNT(DISTINCT pid) FROM owners GROUP BY value "
+    "SELECT value,MIN(pid),COUNT(DISTINCT CASE WHEN shelved THEN pid END) "
+    "FROM owners GROUP BY value "
     "UNION ALL SELECT value," + str(_IMAGE_ONLY_FIRST_PID) + ",0 FROM image_values "
     "WHERE value NOT IN (SELECT value FROM owners)),"
     "counted AS (SELECT g.value,g.first_pid,g.prompt_count,"

py/api/__init__.py

@@ -19,6 +19,7 @@ Split into domain-specific mixins for maintainability:
 - SystemRoutesMixin: System health check, Overview and usage statistics
 - LibraryLocationRoutesMixin: where the library lives, Move library, Keep it here
 - MaintenanceRoutesMixin: reviewed maintenance plans (preview, apply)
+- LibraryDeleteRoutesMixin: the Library's Delete… (archive paired) and Restore
 - WorkflowStorageRoutesMixin: store each workflow once, then Reclaim space
 - DuplicateFilesMixin: duplicate image files as find and removal jobs
 - LoraImportRoutesMixin: the reviewed, staged LoRA reference import job
@@ -42,6 +43,7 @@ from PIL import Image
 from .prompts import PromptRoutesMixin
 from .pairs import PairRoutesMixin
 from .library import LibraryRoutesMixin
+from .library_delete import LibraryDeleteRoutesMixin
 from .vocabulary import VocabularyRoutesMixin
 from .annotations import AnnotationRoutesMixin
 from .review import ReviewRoutesMixin
@@ -313,6 +315,7 @@ async def _gzip_middleware(request, handler):
 class PromptManagerAPI(
     PairRoutesMixin,
     LibraryRoutesMixin,
+    LibraryDeleteRoutesMixin,
     VocabularyRoutesMixin,
     AnnotationRoutesMixin,
     ReviewRoutesMixin,
@@ -687,6 +690,7 @@ class PromptManagerAPI(
 
         self._register_pair_routes(routes)
         self._register_library_routes(routes)
+        self._register_library_delete_routes(routes)
         self._register_vocabulary_routes(routes)
         self._register_annotation_routes(routes)
         self._register_review_routes(routes)

py/api/jobs.py

@@ -123,6 +123,9 @@ MODEL_DOWNLOAD_JOB_KIND = "model_download"
 DUPLICATES_JOB_KIND = "duplicates"
 DUPLICATE_GROUP_NAMES = ("groups_found",)
 DUPLICATE_CLEANUP_GROUP_NAMES = ("removed", "kept", "errors")
+# The workspace Library's Delete… (#443, py/api/library_delete.py): a
+# ``records`` hold around its safety backup and write.
+PROMPT_DELETE_JOB_KIND = "prompt_delete"
 # 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"

py/api/library.py

@@ -155,6 +155,13 @@ def _archive_scope(request: web.Request) -> str:
     return value
 
 
+def _shelf(request: web.Request) -> str:
+    value = request.query.get("shelf", "active")
+    if value not in ("active", "archived"):
+        raise ValueError("shelf must be active or archived")
+    return value
+
+
 def _missing() -> web.Response:
     return web.json_response(
         {"success": False, "error": "Positive prompt not found"}, status=404
@@ -295,6 +302,9 @@ class LibraryRoutesMixin:
             "category": request.query.get("category"),
             "tags": sorted(set(request.query.getall("tag", []))),
             "sort": request.query.get("sort", "newest"),
+            # #443: which shelf; part of the cursor scope, so a cursor never
+            # crosses shelves. Not the variant ``archive`` scope of /{id}.
+            "shelf": _shelf(request),
         }
         limit = _page_limit(request)
         cursor = request.query.get("cursor")

py/api/library_delete.py

@@ -0,0 +1,263 @@
+"""Delete or archive selected Library prompts, and restore archived ones (#443).
+
+The workspace Prompt library's Delete… (D152). The legacy ``POST
+/prompt_manager/bulk/delete`` and ``DELETE /prompt_manager/delete/{id}`` are
+separate and unchanged.
+
+- ``POST /prompt_manager/library/delete/preview {prompt_ids, expected_epoch}``
+  classifies the selection (reads only, under a lease): the ids to ``delete``
+  (in no saved pair), to ``archive`` (paired), the ``protected`` ones (tagged
+  ``__protected__``, kept), counts of ``already_archived`` and ``not_found``,
+  the ``image_links`` a delete removes, the safety ``backup`` policy and,
+  when a backup would be taken, the ``disk`` it needs.
+- ``POST /prompt_manager/library/delete {archive_ids, delete_ids,
+  expected_epoch, backup}`` executes that review under a ``records`` hold
+  (refused 409 ``busy`` beside a scan, restore, maintenance, move or
+  Optimize). When a delete remains and a backup was promised (the reviewed
+  ``backup`` was "on" or the policy is on now) it checks the disk (507
+  ``insufficient-space``) and takes a verified safety backup (500
+  ``backup-failed``); then one transaction deletes the reviewed deletes whose
+  fingerprint is unchanged since the backup and archives the paired ones.
+- ``POST /prompt_manager/library/restore {prompt_ids, expected_epoch}``
+  removes the archive mark (under a lease).
+
+A changed storage epoch answers 409 ``storage-changed``; nothing is written.
+"""
+
+import sqlite3
+
+from aiohttp import web
+
+try:
+    from ...database.library_commands import LibraryCommandError
+    from ...database.safety_backup import SafetyBackupFailed
+except ImportError:
+    from database.library_commands import LibraryCommandError
+    from database.safety_backup import SafetyBackupFailed
+
+from . import maintenance as _maintenance
+from .admin import _prune_safety_backups
+from .job_routes import busy_response
+from .jobs import PROMPT_DELETE_JOB_KIND, JobBusy
+from .prompts import parse_bulk_prompt_ids
+
+PREVIEW_URL = "/prompt_manager/library/delete/preview"
+DELETE_URL = "/prompt_manager/library/delete"
+RESTORE_URL = "/prompt_manager/library/restore"
+DELETE_TITLE = "Delete prompts (Library)"
+_SELECTION_KEYS = frozenset({"prompt_ids", "expected_epoch"})
+_DELETE_KEYS = frozenset({"archive_ids", "delete_ids", "expected_epoch", "backup"})
+STORAGE_CHANGED = "Prompt storage was restored; refresh the library"
+
+
+def _failure(status, code, error, **extra):
+    return web.json_response(
+        {"success": False, "code": code, "error": error, **extra}, status=status
+    )
+
+
+def _invalid(message):
+    return _failure(400, "invalid-request", message)
+
+
+def _storage_changed(exc):
+    return _failure(409, exc.code, STORAGE_CHANGED, library_epoch=exc.epoch)
+
+
+async def _body(request, keys):
+    """The JSON object with exactly ``keys``, or raise ValueError."""
+    try:
+        data = await request.json()
+    except ValueError:
+        raise ValueError("Expected a JSON object") from None
+    if not isinstance(data, dict):
+        raise ValueError("Expected a JSON object")
+    if set(data) != keys:
+        raise ValueError("Request must contain exactly " + ", ".join(sorted(keys)))
+    return data
+
+
+def _reviewed_ids(value, name):
+    """A reviewed id list: possibly empty, else as ``parse_bulk_prompt_ids``."""
+    if value == []:
+        return []
+    try:
+        return parse_bulk_prompt_ids(value)
+    except ValueError as exc:
+        raise ValueError(str(exc).replace("prompt_ids", name)) from None
+
+
+class LibraryDeleteRoutesMixin:
+    """Mixin providing the workspace Library's delete, archive and restore."""
+
+    def _register_library_delete_routes(self, routes):
+        routes.post(PREVIEW_URL)(self.preview_library_delete)
+        routes.post(DELETE_URL)(self.delete_library_prompts)
+        routes.post(RESTORE_URL)(self.restore_library_prompts)
+
+    async def preview_library_delete(self, request):
+        try:
+            data = await _body(request, _SELECTION_KEYS)
+            ids = parse_bulk_prompt_ids(data["prompt_ids"])
+        except ValueError as exc:
+            return _invalid(str(exc))
+        db = self.db
+        policy = self._safety_backup_policy()
+        try:
+            async with self._jobs.lease("Library delete review"):
+                review = await self._run_in_executor(
+                    db.classify_library_delete,
+                    ids,
+                    expected_epoch=data["expected_epoch"],
+                )
+                disk = None
+                if review["delete"] and policy == "on":
+                    public, short, _size = await self._run_in_executor(
+                        _maintenance.disk_plan,
+                        db.model.db_path,
+                        backup=True,
+                        vacuum=False,
+                    )
+                    disk = {
+                        **public,
+                        "message": (
+                            None
+                            if short is None
+                            else _maintenance.short_space_message(short)
+                        ),
+                    }
+        except JobBusy as busy:
+            return busy_response(busy)
+        except LibraryCommandError as exc:
+            return _storage_changed(exc)
+        except ValueError as exc:
+            return _invalid(str(exc))
+        except (OSError, sqlite3.Error):
+            self.logger.exception("Library delete review failed")
+            return _failure(500, "review-failed", "The delete could not be reviewed")
+        return web.json_response(
+            {
+                "success": True,
+                "library_epoch": review["library_epoch"],
+                "delete": review["delete"],
+                "archive": review["archive"],
+                "protected": review["protected"],
+                "already_archived": len(review["already_archived"]),
+                "not_found": len(review["not_found"]),
+                "image_links": review["image_links"],
+                "backup": policy,
+                "disk": disk,
+            }
+        )
+
+    async def delete_library_prompts(self, request):
+        try:
+            data = await _body(request, _DELETE_KEYS)
+            archive_ids = _reviewed_ids(data["archive_ids"], "archive_ids")
+            delete_ids = _reviewed_ids(data["delete_ids"], "delete_ids")
+            if not archive_ids and not delete_ids:
+                raise ValueError("Nothing to delete or archive")
+            if len(archive_ids) + len(delete_ids) > 1000:
+                raise ValueError("At most 1000 prompts can be changed at once")
+            if set(archive_ids) & set(delete_ids):
+                raise ValueError("A prompt cannot be both archived and deleted")
+            if data["backup"] not in ("on", "off"):
+                raise ValueError("backup must be on or off")
+        except ValueError as exc:
+            return _invalid(str(exc))
+        busy = self._jobs.blocker(PROMPT_DELETE_JOB_KIND, admission="records")
+        if busy is not None:
+            return busy_response(JobBusy(busy))
+        try:
+            async with self._jobs.hold(
+                PROMPT_DELETE_JOB_KIND, DELETE_TITLE, admission="records"
+            ) as job:
+                response = await self._held_library_delete(
+                    self.db, data, archive_ids, delete_ids
+                )
+                if response.status >= 400:
+                    job.finish("failed")
+                return response
+        except JobBusy as busy:
+            return busy_response(busy)
+
+    async def _held_library_delete(self, db, data, archive_ids, delete_ids):
+        epoch = data["expected_epoch"]
+        promised = data["backup"] == "on" or self._safety_backup_policy() == "on"
+        backup = None
+        try:
+            candidates = None
+            if delete_ids:
+                review = await self._run_in_executor(
+                    db.classify_library_delete, delete_ids, expected_epoch=epoch
+                )
+                candidates = review["fingerprints"]
+                if candidates and promised:
+                    _disk, short, _size = await self._run_in_executor(
+                        _maintenance.disk_plan,
+                        db.model.db_path,
+                        backup=True,
+                        vacuum=False,
+                    )
+                    if short is not None:
+                        return _failure(
+                            507,
+                            "insufficient-space",
+                            _maintenance.short_space_message(short),
+                        )
+                    backup = await self._run_in_executor(
+                        _maintenance.take_safety_backup, db.model
+                    )
+            result = await self._run_in_executor(
+                db.archive_or_delete_library_prompts,
+                archive_ids,
+                delete_ids,
+                expected_epoch=epoch,
+                candidates=candidates if candidates is not None else {},
+            )
+        except LibraryCommandError as exc:
+            return _storage_changed(exc)
+        except SafetyBackupFailed as error:
+            self.logger.error("Safety backup before a Library delete failed")
+            return _failure(
+                500, "backup-failed", _maintenance.backup_failed_message(error)
+            )
+        except ValueError as exc:
+            return _invalid(str(exc))
+        except (OSError, sqlite3.Error):
+            self.logger.exception("Library delete failed")
+            return _failure(500, "delete-failed", "The prompts could not be changed")
+        finally:
+            if backup is not None:
+                await self._prune_after_delete(db)
+        return web.json_response({"success": True, **result, "backup": backup})
+
+    async def _prune_after_delete(self, db):
+        try:
+            await self._run_in_executor(_prune_safety_backups, db.model.db_path)
+        except Exception:  # pruning old copies never fails the delete
+            self.logger.exception("Could not prune old safety backups")
+
+    async def restore_library_prompts(self, request):
+        try:
+            data = await _body(request, _SELECTION_KEYS)
+            ids = parse_bulk_prompt_ids(data["prompt_ids"])
+        except ValueError as exc:
+            return _invalid(str(exc))
+        try:
+            async with self._jobs.lease("Library restore"):
+                result = await self._run_in_executor(
+                    self.db.restore_library_prompts,
+                    ids,
+                    expected_epoch=data["expected_epoch"],
+                )
+        except JobBusy as busy:
+            return busy_response(busy)
+        except LibraryCommandError as exc:
+            return _storage_changed(exc)
+        except ValueError as exc:
+            return _invalid(str(exc))
+        except (OSError, sqlite3.Error):
+            self.logger.exception("Library restore failed")
+            return _failure(500, "restore-failed", "The prompts could not be restored")
+        return web.json_response({"success": True, **result})

tests/test_bulk_library_delete.py

@@ -0,0 +1,484 @@
+"""#443 (D152): delete selected Library prompts; paired ones are archived.
+
+The DB commands, and what every reader does with an archived parent.
+"""
+
+import re
+import sqlite3
+import tempfile
+import unittest
+from pathlib import Path
+
+from database.execution_history_schema import rotate_history_epoch
+from database.library_commands import LibraryCommandError
+from database.operations import PromptDatabase
+from database.prompt_archive import TABLE as ARCHIVE
+from utils.hashing import generate_prompt_hash
+
+
+def save(db, text, **metadata):
+    return db.save_prompt(text, prompt_hash=generate_prompt_hash(text), **metadata)
+
+
+def rows(db, sql, params=()):
+    with db.model.get_connection() as conn:
+        return [tuple(row) for row in conn.execute(sql, params).fetchall()]
+
+
+def archived_ids(db):
+    return sorted(row[0] for row in rows(db, f"SELECT prompt_id FROM {ARCHIVE}"))
+
+
+def prompt_ids(db):
+    return sorted(row[0] for row in rows(db, "SELECT id FROM prompts"))
+
+
+_STAMP = re.compile(r"^\d{4}-\d{2}-\d{2}[T ]\d{2}:\d{2}:\d{2}")
+
+
+def _stable(value):
+    return "<time>" if isinstance(value, str) and _STAMP.match(value) else value
+
+
+def table_dump(db):
+    """Every row of every table (times masked), to compare two databases."""
+    tables = [
+        row[0]
+        for row in rows(
+            db,
+            "SELECT name FROM sqlite_master WHERE type='table' "
+            "AND name NOT LIKE 'sqlite_%' AND name!='execution_history_metadata' "
+            "ORDER BY name",
+        )
+    ]
+    dump = {}
+    for table in tables:
+        dump[table] = sorted(
+            (tuple(map(_stable, row)) for row in rows(db, f"SELECT * FROM {table}")),
+            key=repr,
+        )
+    return dump
+
+
+class LibraryDeleteCase(unittest.TestCase):
+    def setUp(self):
+        self.temp = tempfile.TemporaryDirectory()
+        self.root = Path(self.temp.name)
+        self.db = PromptDatabase(str(self.root / "delete.db"))
+        self.epoch = self.db.get_library_options()["library_epoch"]
+
+    def tearDown(self):
+        self.db.close_all()
+        self.temp.cleanup()
+
+    def paired(self, positive, negative="blur"):
+        pair = self.db.save_prompt_pair(positive, negative)
+        found = self.db.get_prompt_pair(pair)
+        return found["positive_prompt_id"], found["negative_prompt_id"], pair
+
+    def image(self, prompt_id, name):
+        path = self.root / name
+        path.write_bytes(b"png")
+        return self.db.link_image_to_prompt(prompt_id, str(path))
+
+    def apply(self, ids, epoch=None, review=None):
+        """Review ``ids`` (or use ``review``), then write what was reviewed."""
+        review = review or self.classify(ids)
+        return self.db.archive_or_delete_library_prompts(
+            review["archive"],
+            review["delete"],
+            expected_epoch=epoch or self.epoch,
+            candidates=review["fingerprints"],
+        )
+
+    def classify(self, ids, epoch=None):
+        return self.db.classify_library_delete(ids, expected_epoch=epoch or self.epoch)
+
+    def restore(self, ids, epoch=None):
+        return self.db.restore_library_prompts(ids, expected_epoch=epoch or self.epoch)
+
+
+class TestClassifyAndApply(LibraryDeleteCase):
+    def test_each_class_in_order(self):
+        plain = save(self.db, "Plain")
+        protected = save(self.db, "Guarded", tags=["__protected__"])
+        protected_paired, _neg, _pair = self.paired("Guarded pair")
+        self.db.bulk_add_tags([protected_paired], ["__protected__"])
+        positive, negative_only, _pair = self.paired("Cat", "only negative")
+        both = save(self.db, "Both roles")
+        self.image(both, "both.png")  # a parent through its image
+        self.db.save_prompt_pair("Other", "Both roles")
+        archived_pair_positive, _n, old_pair = self.paired("Old")
+        self.db.archive_prompt_pair(old_pair, True)
+        already, _n2, _p2 = self.paired("Already")
+        self.apply([already])
+        self.image(plain, "plain.png")
+        ids = [
+            plain,
+            protected,
+            protected_paired,
+            positive,
+            negative_only,
+            both,
+            archived_pair_positive,
+            already,
+            99999,
+        ]
+        result = self.classify(ids)
+        self.assertEqual(
+            result,
+            {
+                "library_epoch": self.epoch,
+                "delete": [plain],
+                "archive": [positive, both, archived_pair_positive],
+                "protected": [protected, protected_paired],
+                "already_archived": [already],
+                "not_found": [negative_only, 99999],
+                "image_links": 1,
+                "fingerprints": result["fingerprints"],
+            },
+        )
+        self.assertEqual(list(result["fingerprints"]), [plain])
+
+    def test_apply_archives_paired_deletes_unpaired_and_skips_protected(self):
+        plain = save(self.db, "Plain", tags=["red"])
+        protected = save(self.db, "Guarded", tags=["__protected__"])
+        positive, negative, pair = self.paired("Cat")
+        self.image(plain, "a.png")
+        self.image(positive, "b.png")
+        before = rows(self.db, "SELECT updated_at FROM prompts WHERE id=?", (positive,))
+        result = self.apply([plain, protected, positive, 99999])
+        self.assertEqual(
+            result,
+            {
+                "library_epoch": self.epoch,
+                "deleted": 1,
+                "archived": 1,
+                "already_archived": 0,
+                "not_found": 0,
+                "changed": 0,
+                "image_links": 1,
+            },
+        )
+        self.assertEqual(prompt_ids(self.db), sorted([protected, positive, negative]))
+        self.assertEqual(archived_ids(self.db), [positive])
+        # Archive keeps everything: the pair, its image link and the prompt row.
+        self.assertEqual(
+            len(rows(self.db, "SELECT 1 FROM prompt_pairs WHERE id=?", (pair,))), 1
+        )
+        self.assertEqual(
+            len(
+                rows(
+                    self.db,
+                    "SELECT 1 FROM generated_images WHERE prompt_id=?",
+                    (positive,),
+                )
+            ),
+            1,
+        )
+        after = rows(self.db, "SELECT updated_at FROM prompts WHERE id=?", (positive,))
+        self.assertNotEqual(before, after)
+        # The image file stays on disk.
+        self.assertTrue((self.root / "a.png").exists())
+
+    def test_delete_has_exactly_the_legacy_row_effects(self):
+        twin = PromptDatabase(str(self.root / "twin.db"))
+        self.addCleanup(twin.close_all)
+        twin_epoch = twin.get_library_options()["library_epoch"]
+        for db in (self.db, twin):
+            a = save(db, "Alpha", tags=["x", "y"], category="c")
+            save(db, "Beta")
+            path = self.root / "shared.png"
+            path.write_bytes(b"png")
+            db.link_image_to_prompt(a, str(path), {"workflow": {"nodes": []}})
+        self.db.bulk_delete_prompts([a])
+        twin.archive_or_delete_library_prompts(
+            [], [a], expected_epoch=twin_epoch, candidates=None
+        )
+        self.assertEqual(table_dump(self.db), table_dump(twin))
+
+    def test_already_archived_and_restore(self):
+        positive, _neg, _pair = self.paired("Cat")
+        self.apply([positive])
+        again = self.classify([positive])
+        self.assertEqual(
+            (again["archive"], again["already_archived"]), ([], [positive])
+        )
+        # Archived between the review and the write: counted, not archived twice.
+        other, _n, _p = self.paired("Dog")
+        review = self.classify([other])
+        self.apply([other])
+        late = self.apply([], review=review)
+        self.assertEqual((late["archived"], late["already_archived"]), (0, 1))
+        plain = save(self.db, "Plain")
+        result = self.restore([positive, plain, 99999])
+        self.assertEqual(
+            result,
+            {
+                "library_epoch": self.epoch,
+                "restored": 1,
+                "unchanged": 1,
+                "not_found": 1,
+            },
+        )
+        self.assertEqual(archived_ids(self.db), [other])
+
+    def test_the_write_executes_only_what_was_reviewed(self):
+        plain = save(self.db, "Plain")
+        guarded = save(self.db, "Guarded later")
+        touched = save(self.db, "Touched later")
+        review = self.classify([plain, guarded, touched])
+        # Between the review and the write: one becomes protected, one is used.
+        self.db.bulk_add_tags([guarded], ["__protected__"])
+        self.db.record_prompt_use(touched)
+        result = self.apply([], review=review)
+        self.assertEqual((result["deleted"], result["changed"]), (1, 2))
+        self.assertEqual(prompt_ids(self.db), sorted([guarded, touched]))
+        self.assertEqual(archived_ids(self.db), [])
+
+    def test_a_new_image_link_after_the_review_keeps_the_prompt(self):
+        plain = save(self.db, "Plain")
+        review = self.classify([plain])
+        self.image(plain, "late.png")
+        self.assertEqual(self.apply([], review=review)["changed"], 1)
+        self.assertIn(plain, prompt_ids(self.db))
+
+    def test_reviewed_deletes_without_a_backup_are_kept(self):
+        plain = save(self.db, "Plain")
+        review = self.classify([plain])
+        result = self.db.archive_or_delete_library_prompts(
+            [], [plain], expected_epoch=self.epoch, candidates={}
+        )
+        self.assertEqual((result["deleted"], result["changed"]), (0, 1))
+        self.assertEqual(review["delete"], [plain])
+
+    def test_a_reviewed_delete_that_became_paired_is_archived(self):
+        plain = save(self.db, "Plain")
+        review = self.classify([plain])
+        self.db.save_prompt_pair("Plain", "late negative")
+        result = self.apply([], review=review)
+        self.assertEqual((result["deleted"], result["archived"]), (0, 1))
+
+    def test_a_reviewed_archive_that_lost_its_pair_is_archived_not_deleted(self):
+        positive, _negative, pair = self.paired("Cat")
+        review = self.classify([positive])
+        self.db.delete_prompt_pair(pair)
+        result = self.apply([], review=review)
+        self.assertEqual((result["deleted"], result["archived"]), (0, 1))
+        self.assertEqual(archived_ids(self.db), [positive])
+
+    def test_one_id_in_both_lists_or_no_ids_is_refused(self):
+        plain = save(self.db, "Plain")
+        for archive, delete in (([plain], [plain]), ([], [])):
+            with self.subTest(archive=archive, delete=delete):
+                with self.assertRaises(ValueError):
+                    self.db.archive_or_delete_library_prompts(
+                        archive, delete, expected_epoch=self.epoch, candidates=None
+                    )
+
+    def test_archiving_repairs_a_missing_or_stale_hash(self):
+        missing, _n1, _p1 = self.paired("Hashless")
+        stale, _n2, _p2 = self.paired("Stale hash")
+        with self.db.model.get_connection() as conn:
+            conn.execute("UPDATE prompts SET hash=NULL WHERE id=?", (missing,))
+            conn.execute("UPDATE prompts SET hash='stale' WHERE id=?", (stale,))
+        self.apply([missing, stale])
+        for pid, text in ((missing, "Hashless"), (stale, "Stale hash")):
+            self.assertEqual(
+                rows(self.db, "SELECT hash FROM prompts WHERE id=?", (pid,)),
+                [(generate_prompt_hash(text),)],
+            )
+            # So every ingestion lookup finds the archived row.
+            self.assertEqual(save(self.db, text), pid)
+
+    def test_a_canonical_hash_another_row_holds_is_left_alone(self):
+        stale, _n, _p = self.paired("Twin text")
+        with self.db.model.get_connection() as conn:
+            conn.execute("UPDATE prompts SET hash='stale' WHERE id=?", (stale,))
+        twin = save(self.db, "twin text")
+        self.apply([stale])
+        self.assertEqual(
+            rows(self.db, "SELECT hash FROM prompts WHERE id=?", (stale,)), [("stale",)]
+        )
+        self.assertEqual(save(self.db, "Twin text"), twin)
+
+    def test_restore_unhides_protected_and_non_parent_archived_prompts(self):
+        guarded, _n, _p = self.paired("Guarded")
+        self.apply([guarded])
+        self.db.bulk_add_tags([guarded], ["__protected__"])
+        negative_only = save(self.db, "Only a negative")
+        self.db.save_prompt_pair("Owner", "Only a negative")
+        with self.db.model.get_connection() as conn:
+            conn.execute(
+                f"INSERT INTO {ARCHIVE} VALUES (?, '2026-10-28T00:00:00+00:00')",
+                (negative_only,),
+            )
+        result = self.restore([guarded, negative_only])
+        self.assertEqual(result["restored"], 2)
+        self.assertEqual(archived_ids(self.db), [])
+
+    def test_restored_storage_changes_nothing(self):
+        plain = save(self.db, "Plain")
+        positive, _neg, _pair = self.paired("Cat")
+        rotate_history_epoch(self.db.model.get_connection())
+        for call in (
+            lambda: self.apply([plain, positive]),
+            lambda: self.classify([plain]),
+            lambda: self.restore([positive]),
+        ):
+            with self.assertRaises(LibraryCommandError) as caught:
+                call()
+            self.assertEqual(caught.exception.code, "storage-changed")
+        self.assertIn(plain, prompt_ids(self.db))
+        self.assertEqual(archived_ids(self.db), [])
+
+    def test_a_failure_mid_write_rolls_everything_back(self):
+        plain = save(self.db, "Plain")
+        positive, _neg, _pair = self.paired("Cat")
+        with self.db.model.get_connection() as conn:
+            conn.execute(
+                "CREATE TRIGGER fail_delete BEFORE DELETE ON prompts "
+                "BEGIN SELECT RAISE(ABORT, 'boom'); END"
+            )
+        with self.assertRaises(sqlite3.DatabaseError):
+            self.apply([positive, plain])
+        self.assertIn(plain, prompt_ids(self.db))
+        self.assertEqual(archived_ids(self.db), [])
+
+    def test_many_ids_are_handled_in_batches(self):
+        ids = [save(self.db, f"Many {i}") for i in range(1000)]
+        result = self.apply(ids)
+        self.assertEqual(result["deleted"], 1000)
+
+    def test_rejects_invalid_arguments(self):
+        plain = save(self.db, "Plain")
+        for ids, epoch in (([], self.epoch), ([True], self.epoch), ([plain], "x")):
+            for call in (self.apply, self.classify, self.restore):
+                with self.subTest(ids=ids, epoch=epoch, call=call.__name__):
+                    with self.assertRaises(ValueError):
+                        call(ids, epoch)
+
+
+class TestArchivedReaders(LibraryDeleteCase):
+    """What each reader does with an archived parent (design §4.4)."""
+
+    def setUp(self):
+        super().setUp()
+        self.kept, _kn, _kp = self.paired("Kept cat", "kept blur")
+        self.gone, _gn, _gp = self.paired("Gone cat", "gone blur")
+        for pid, category, tag in ((self.kept, "keep", "k"), (self.gone, "gone", "g")):
+            self.db.update_prompt_metadata(pid, category=category, tags=[tag])
+        self.apply([self.gone])
+
+    def ids_of(self, page):
+        return [item["id"] for item in page["items"]]
+
+    def test_listing_search_and_totals(self):
+        active = self.db.list_library()
+        self.assertEqual(self.ids_of(active), [self.kept])
+        self.assertEqual(
+            (active["total"], active["variant_total"], active["archived_total"]),
+            (1, 1, 1),
+        )
+        archived = self.db.list_library(shelf="archived")
+        self.assertEqual(self.ids_of(archived), [self.gone])
+        self.assertEqual((archived["total"], archived["variant_total"]), (1, 1))
+        self.assertEqual(self.ids_of(self.db.list_library(text="Gone")), [])
+        self.assertEqual(
+            self.ids_of(self.db.list_library(text="Gone", shelf="archived")),
+            [self.gone],
+        )
+        self.assertEqual(self.ids_of(self.db.list_library(tags=["g"])), [])
+        self.assertEqual(self.db.list_library(category="gone")["archived_total"], 1)
+        with self.assertRaises(ValueError):
+            self.db.list_library(shelf="all")
+
+    def test_detail_and_options(self):
+        self.assertTrue(self.db.get_library_positive(self.gone)["positive"]["archived"])
+        self.assertFalse(
+            self.db.get_library_positive(self.kept)["positive"]["archived"]
+        )
+        # Suggestions keep archived parents' categories: the archived view
+        # filters by them.
+        self.assertEqual(self.db.get_library_options()["categories"], ["gone", "keep"])
+
+    def test_counts(self):
+        self.assertEqual(self.db.library_counts()["positive_prompts"], 1)
+        self.assertEqual(self.db.library_counts()["negative_variants"], 1)
+        tags = {
+            row["name"]: row["prompt_count"]
+            for row in self.db.list_vocabulary_tags()["tags"]
+        }
+        self.assertEqual((tags["k"], tags["g"]), (1, 0))
+        categories = {
+            row["value"]: row["prompt_count"]
+            for row in self.db.list_vocabulary_categories()["categories"]
+        }
+        # A value only archived parents hold stays listed with 0, as a tag does.
+        self.assertEqual(categories, {"keep": 1, "gone": 0})
+
+    def test_restore_summary_and_export(self):
+        from database.restore_stage import library_summary
+
+        summary = library_summary(self.db.model.get_connection())
+        self.assertEqual(summary["positives"], 1)
+        exported = {
+            item["positive_text"]: item["archived"]
+            for item in self.db.export_library()["positives"]
+        }
+        self.assertEqual(exported, {"Kept cat": False, "Gone cat": True})
+
+    def test_usage_statistics_keep_archived_runs(self):
+        self.db.record_prompt_use(self.gone, 3)
+        runs = {row["category"]: row["runs"] for row in self.db.usage_by_category()}
+        self.assertEqual(runs["gone"], 3)
+
+    def test_runs_node_saves_and_new_prompt_find_the_archived_row(self):
+        count = len(prompt_ids(self.db))
+        self.assertTrue(self.db.record_prompt_use(self.gone, 2))
+        existing = self.db.get_prompt_by_hash(generate_prompt_hash("Gone cat"))
+        self.assertEqual(existing["id"], self.gone)
+        self.assertEqual(save(self.db, "Gone cat"), self.gone)
+        self.db.save_prompt_pair("Gone cat", "new blur")
+        self.assertEqual(len(prompt_ids(self.db)), count + 1)  # only the new negative
+        self.assertEqual(archived_ids(self.db), [self.gone])
+
+    def test_legacy_reads_still_list_it_and_legacy_delete_still_refuses(self):
+        from database.pairs import PairDependencyError
+
+        texts = [p["text"] for p in self.db.get_recent_prompts(limit=50)["prompts"]]
+        self.assertIn("Gone cat", texts)
+        with self.assertRaises(PairDependencyError):
+            self.db.bulk_delete_prompts([self.gone])
+
+    def test_as_another_parents_negative_it_stays_a_visible_variant(self):
+        owner, _neg, pair = self.paired("Owner cat", "Gone cat")
+        detail = self.db.get_library_positive(owner)
+        self.assertEqual([v["id"] for v in detail["variants"]], [pair])
+        self.assertEqual(self.ids_of(self.db.list_library(text="Gone")), [owner])
+
+    def test_maintenance_never_prunes_an_archived_prompt(self):
+        lonely = save(self.db, "Lonely")
+        with self.db.model.get_connection() as conn:
+            conn.execute(
+                f"INSERT INTO {ARCHIVE} VALUES (?, '2026-10-28T00:00:00+00:00')",
+                (lonely,),
+            )
+            orphaned = [row[0] for row in conn.execute(self.orphaned_sql()).fetchall()]
+        self.assertNotIn(lonely, orphaned)
+
+    @staticmethod
+    def orphaned_sql():
+        from database.maintenance import ORPHANED_PROMPTS_SQL
+
+        return ORPHANED_PROMPTS_SQL
+
+    def test_bulk_metadata_writes_still_reach_an_archived_parent(self):
+        result = self.db.bulk_set_library_category(
+            [self.gone], "moved", expected_epoch=self.epoch
+        )
+        self.assertEqual(result["updated"], 1)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_history_reads.py

@@ -389,6 +389,32 @@ class HistoryReadTests(unittest.TestCase):
                 pass
         self.assertFalse(conn.in_transaction)
 
+    def test_a_deleted_negative_is_unavailable_and_a_negative_only_one_is_not(self):
+        # #443: a Library delete can remove an unpaired prompt a run used as
+        # its negative. The run's prompt ids are live hints, so a deleted one
+        # reads unavailable; a negative-only prompt stays available.
+        db = self.database()
+        self.saved(db)
+        self.saved(db, "negative", 2)
+        start = self.capture(db)
+        with db.recorded_history_snapshot():
+            context = db.get_history_run(start.run_id)["contexts"][0]
+        self.assertTrue(context["navigation"]["negative_available"])
+        with db.model.get_connection() as conn:
+            conn.execute(
+                "INSERT INTO prompt_pairs(id,positive_prompt_id,negative_prompt_id) "
+                "VALUES(9,1,2)"
+            )
+        with db.recorded_history_snapshot():
+            context = db.get_history_run(start.run_id)["contexts"][0]
+        self.assertTrue(context["navigation"]["negative_available"])
+        with db.model.get_connection() as conn:
+            conn.execute("DELETE FROM prompt_pairs WHERE id=9")
+            conn.execute("DELETE FROM prompts WHERE id=2")
+        with db.recorded_history_snapshot():
+            context = db.get_history_run(start.run_id)["contexts"][0]
+        self.assertFalse(context["navigation"]["negative_available"])
+
     def test_detail_preserves_full_text_snapshots_hints_and_archive(self):
         db = self.database()
         self.saved(db)

tests/test_library.py

@@ -345,7 +345,8 @@ class TestLibrary(unittest.TestCase):
         self.assertEqual(
             (parent["title"], parent["notes"]), ("Portrait", "keep\nnotes")
         )
-        self.assertNotIn("archived", parent)
+        # The parent's own archive state (#443), never a pair's.
+        self.assertIs(parent["archived"], False)
         self.assertRegex(parent["edit_token"], r"^[0-9a-f]{64}$")
         self.assertEqual(page["items"][1]["title"], " \t\u00a0\u3000")
         self.assertEqual(self.db.list_library(text="Portrait")["total"], 1)
@@ -1520,6 +1521,7 @@ class TestLibrary(unittest.TestCase):
                     "tags",
                     "rating",
                     "notes",
+                    "archived",
                     "variants",
                 },
             )

tests/test_library_delete_routes.py

@@ -0,0 +1,353 @@
+"""#443 (D152): the Library delete, archive and restore routes, and ingestion after."""
+
+import json
+import os
+from unittest.mock import patch
+
+from tests.test_api_admin import AdminAPITestCase  # first: it stubs ComfyUI
+from database.execution_history_schema import rotate_history_epoch  # noqa: E402
+from database.prompt_archive import TABLE as ARCHIVE  # noqa: E402
+from database.safety_backup import SafetyBackupFailed  # noqa: E402
+from py.config import PromptManagerConfig  # noqa: E402
+from utils.hashing import generate_prompt_hash  # noqa: E402
+
+PREVIEW = "/prompt_manager/library/delete/preview"
+DELETE = "/prompt_manager/library/delete"
+RESTORE = "/prompt_manager/library/restore"
+
+
+def graph(positive, negative="blur"):
+    return {
+        "1": {"class_type": "CLIPTextEncode", "inputs": {"text": positive}},
+        "2": {"class_type": "CLIPTextEncode", "inputs": {"text": negative}},
+        "3": {
+            "class_type": "KSampler",
+            "inputs": {"positive": ["1", 0], "negative": ["2", 0]},
+        },
+    }
+
+
+class LibraryDeleteRouteCase(AdminAPITestCase):
+    def epoch(self):
+        return self.api.db.get_library_options()["library_epoch"]
+
+    def save(self, text, **metadata):
+        return self.api.db.save_prompt(
+            text, prompt_hash=generate_prompt_hash(text), **metadata
+        )
+
+    def paired(self, positive, negative="blur"):
+        pair = self.api.db.save_prompt_pair(positive, negative)
+        return self.api.db.get_prompt_pair(pair)["positive_prompt_id"]
+
+    def rows(self, sql, params=()):
+        with self.api.db.model.get_connection() as conn:
+            return [tuple(row) for row in conn.execute(sql, params).fetchall()]
+
+    def ids(self):
+        return sorted(row[0] for row in self.rows("SELECT id FROM prompts"))
+
+    def archived(self):
+        return sorted(row[0] for row in self.rows(f"SELECT prompt_id FROM {ARCHIVE}"))
+
+    def backups(self):
+        folder = os.path.dirname(self.api.db.model.db_path)
+        return sorted(name for name in os.listdir(folder) if ".backup_" in name)
+
+    async def post(self, url, body):
+        response = await self.client.request("POST", url, json=body)
+        return response.status, await response.json()
+
+    async def review(self, ids):
+        status, body = await self.post(
+            PREVIEW, {"prompt_ids": ids, "expected_epoch": self.epoch()}
+        )
+        self.assertEqual(status, 200, body)
+        return body
+
+    async def confirm(self, review, backup=None):
+        return await self.post(
+            DELETE,
+            {
+                "archive_ids": review["archive"],
+                "delete_ids": review["delete"],
+                "expected_epoch": review["library_epoch"],
+                "backup": backup or review["backup"],
+            },
+        )
+
+    def policy(self, on):
+        previous = PromptManagerConfig.SAFETY_BACKUP
+        PromptManagerConfig.SAFETY_BACKUP = on
+        self.addCleanup(setattr, PromptManagerConfig, "SAFETY_BACKUP", previous)
+
+
+class TestLibraryDeleteRoutes(LibraryDeleteRouteCase):
+    async def test_preview_then_delete_archives_deletes_and_backs_up(self):
+        self.policy(True)
+        plain = self.save("Plain")
+        guarded = self.save("Guarded", tags=["__protected__"])
+        positive = self.paired("Cat")
+        review = await self.review([plain, guarded, positive, 99999])
+        self.assertEqual(
+            {
+                key: review[key]
+                for key in ("delete", "archive", "protected", "not_found")
+            },
+            {
+                "delete": [plain],
+                "archive": [positive],
+                "protected": [guarded],
+                "not_found": 1,
+            },
+        )
+        self.assertEqual(review["backup"], "on")
+        self.assertTrue(review["disk"]["sufficient"])
+        status, body = await self.confirm(review)
+        self.assertEqual(status, 200, body)
+        self.assertEqual(
+            (body["deleted"], body["archived"], body["changed"]), (1, 1, 0)
+        )
+        self.assertEqual(self.backups(), [body["backup"]])
+        self.assertNotIn(os.sep, body["backup"])
+        self.assertNotIn(plain, self.ids())
+        self.assertEqual(self.archived(), [positive])
+        # The backup still holds the deleted prompt.
+        import sqlite3
+
+        path = os.path.join(os.path.dirname(self.api.db.model.db_path), body["backup"])
+        with sqlite3.connect(path) as copy:
+            self.assertEqual(
+                copy.execute(
+                    "SELECT COUNT(*) FROM prompts WHERE id=?", (plain,)
+                ).fetchone()[0],
+                1,
+            )
+        copy.close()
+        # The hold ended: nothing is listed as running.
+        self.assertEqual(self.api._jobs.running(), [])
+
+    async def test_archive_only_takes_no_backup(self):
+        self.policy(True)
+        positive = self.paired("Cat")
+        review = await self.review([positive])
+        self.assertIsNone(review["disk"])
+        status, body = await self.confirm(review)
+        self.assertEqual((status, body["archived"], body["backup"]), (200, 1, None))
+        self.assertEqual(self.backups(), [])
+
+    async def test_policy_off_takes_none_unless_the_review_promised_one(self):
+        plain = self.save("Plain")
+        other = self.save("Other")
+        self.policy(False)
+        review = await self.review([plain])
+        self.assertEqual((review["backup"], review["disk"]), ("off", None))
+        status, body = await self.confirm(review)
+        self.assertEqual((status, body["deleted"], body["backup"]), (200, 1, None))
+        # Reviewed with the policy on, turned off before the confirm: kept.
+        review = await self.review([other])
+        status, body = await self.confirm(review, backup="on")
+        self.assertEqual(status, 200)
+        self.assertIsNotNone(body["backup"])
+
+    async def test_a_failed_backup_changes_nothing(self):
+        self.policy(True)
+        plain = self.save("Plain")
+        positive = self.paired("Cat")
+        review = await self.review([plain, positive])
+        with patch(
+            "py.api.maintenance.take_safety_backup",
+            side_effect=SafetyBackupFailed("x"),
+        ):
+            status, body = await self.confirm(review)
+        self.assertEqual((status, body["code"]), (500, "backup-failed"))
+        self.assertIn("nothing was changed", body["error"])
+        self.assertIn(plain, self.ids())
+        self.assertEqual(self.archived(), [])
+
+    async def test_a_short_disk_refuses_before_any_change(self):
+        self.policy(True)
+        plain = self.save("Plain")
+        from py.api import maintenance
+
+        with patch.object(maintenance, "_free_bytes", return_value=0):
+            review = await self.review([plain])
+            self.assertFalse(review["disk"]["sufficient"])
+            self.assertIn("Nothing was changed", review["disk"]["message"])
+            status, body = await self.confirm(review)
+        self.assertEqual((status, body["code"]), (507, "insufficient-space"))
+        self.assertIn(plain, self.ids())
+
+    async def test_busy_beside_a_record_changing_job(self):
+        plain = self.save("Plain")
+        review = await self.review([plain])
+        async with self.api._jobs.hold("scan", "Scan", admission="records"):
+            status, body = await self.confirm(review)
+        self.assertEqual((status, body["code"]), (409, "busy"))
+        self.assertEqual(body["running"]["title"], "Scan")
+        self.assertIn(plain, self.ids())
+
+    async def test_restored_storage_answers_409_on_every_route(self):
+        plain = self.save("Plain")
+        review = await self.review([plain])
+        rotate_history_epoch(self.api.db.model.get_connection())
+        status, body = await self.confirm(review)
+        self.assertEqual((status, body["code"]), (409, "storage-changed"))
+        for url in (PREVIEW, RESTORE):
+            status, body = await self.post(
+                url, {"prompt_ids": [plain], "expected_epoch": review["library_epoch"]}
+            )
+            self.assertEqual((status, body["code"]), (409, "storage-changed"))
+        self.assertIn(plain, self.ids())
+
+    async def test_restore_route(self):
+        positive = self.paired("Cat")
+        await self.confirm(await self.review([positive]))
+        status, body = await self.post(
+            RESTORE, {"prompt_ids": [positive, 99999], "expected_epoch": self.epoch()}
+        )
+        self.assertEqual(status, 200)
+        self.assertEqual((body["restored"], body["not_found"]), (1, 1))
+        self.assertEqual(self.archived(), [])
+
+    async def test_bodies_are_validated(self):
+        plain = self.save("Plain")
+        epoch = self.epoch()
+        selection = {"prompt_ids": [plain], "expected_epoch": epoch}
+        for url, body in (
+            (PREVIEW, {**selection, "extra": 1}),
+            (PREVIEW, {"prompt_ids": [], "expected_epoch": epoch}),
+            (PREVIEW, {"prompt_ids": [True], "expected_epoch": epoch}),
+            (PREVIEW, {"prompt_ids": list(range(1, 1002)), "expected_epoch": epoch}),
+            (PREVIEW, {"prompt_ids": [plain], "expected_epoch": "nope"}),
+            (RESTORE, {"prompt_ids": [plain]}),
+            (DELETE, {**selection}),
+            (
+                DELETE,
+                {
+                    "archive_ids": [],
+                    "delete_ids": [],
+                    "expected_epoch": epoch,
+                    "backup": "on",
+                },
+            ),
+            (
+                DELETE,
+                {
+                    "archive_ids": [plain],
+                    "delete_ids": [plain],
+                    "expected_epoch": epoch,
+                    "backup": "on",
+                },
+            ),
+            (
+                DELETE,
+                {
+                    "archive_ids": [],
+                    "delete_ids": [plain],
+                    "expected_epoch": epoch,
+                    "backup": "maybe",
+                },
+            ),
+            (
+                DELETE,
+                {
+                    "archive_ids": list(range(1, 600)),
+                    "delete_ids": list(range(600, 1100)),
+                    "expected_epoch": epoch,
+                    "backup": "off",
+                },
+            ),
+        ):
+            with self.subTest(url=url, body=body):
+                status, answer = await self.post(url, body)
+                self.assertEqual(status, 400, answer)
+                self.assertFalse(answer["success"])
+        for url in (PREVIEW, DELETE, RESTORE):
+            response = await self.client.request(
+                "POST", url, data="[1]", headers={"Content-Type": "application/json"}
+            )
+            self.assertEqual(response.status, 400)
+        self.assertIn(plain, self.ids())
+
+    async def test_the_legacy_bulk_delete_is_unchanged(self):
+        plain = self.save("Plain")
+        positive = self.paired("Cat")
+        response = await self.client.request(
+            "POST", "/prompt_manager/bulk/delete", json={"prompt_ids": [positive]}
+        )
+        self.assertEqual(response.status, 409)
+        response = await self.client.request(
+            "POST", "/prompt_manager/bulk/delete", json={"prompt_ids": [plain]}
+        )
+        self.assertEqual((await response.json())["deleted_count"], 1)
+        self.assertEqual(self.archived(), [])
+
+    async def test_shelf_parameter_and_cursor_scope(self):
+        kept = [self.paired(f"Kept {i}") for i in range(3)]
+        gone = [self.paired(f"Gone {i}") for i in range(3)]
+        await self.confirm(await self.review(gone))
+        response = await self.client.request(
+            "GET", "/prompt_manager/library?limit=2&shelf=active"
+        )
+        page = await response.json()
+        self.assertEqual(page["archived_total"], 3)
+        self.assertTrue(set(item["id"] for item in page["items"]) <= set(kept))
+        response = await self.client.request(
+            "GET",
+            "/prompt_manager/library?limit=2&shelf=archived"
+            f"&cursor={page['next_cursor']}",
+        )
+        self.assertEqual(response.status, 400)
+        response = await self.client.request(
+            "GET", "/prompt_manager/library?shelf=archived"
+        )
+        archived = await response.json()
+        self.assertEqual(sorted(item["id"] for item in archived["items"]), gone)
+        self.assertTrue(all(item["archived"] for item in archived["items"]))
+        response = await self.client.request("GET", "/prompt_manager/library?shelf=all")
+        self.assertEqual(response.status, 400)
+
+
+class TestIngestionAfterArchive(LibraryDeleteRouteCase):
+    def scan(self, name, positive, negative="blur"):
+        meta = {"prompt": json.dumps(graph(positive, negative))}
+        path = self.output_dir / name
+        path.write_bytes(b"png")
+        return self.api._ingest_scanned_file(path, meta)
+
+    async def test_a_scan_after_archive_creates_no_duplicate_and_keeps_it_archived(
+        self,
+    ):
+        self.assertEqual(self.scan("first.png", "Sunset"), "added")
+        positive = self.api.db.get_prompt_by_hash(generate_prompt_hash("Sunset"))["id"]
+        await self.confirm(await self.review([positive]))
+        before = self.ids()
+        self.scan("second.png", "Sunset")
+        self.scan("third.png", "  sunset ", "other negative")
+        self.assertEqual(
+            [i for i in self.ids() if i not in before],
+            [
+                self.api.db.get_prompt_by_hash(generate_prompt_hash("other negative"))[
+                    "id"
+                ]
+            ],
+        )
+        self.assertEqual(self.archived(), [positive])
+        linked = self.rows(
+            "SELECT COUNT(*) FROM generated_images WHERE prompt_id=?", (positive,)
+        )
+        self.assertEqual(linked, [(3,)])
+        response = await self.client.request("GET", "/prompt_manager/library")
+        self.assertNotIn(
+            positive, [item["id"] for item in (await response.json())["items"]]
+        )
+
+    async def test_a_scan_after_delete_creates_the_prompt_again(self):
+        plain = self.save("Lonely sunset")
+        await self.confirm(await self.review([plain]))
+        self.assertEqual(self.scan("again.png", "Lonely sunset"), "added")
+        again = self.api.db.get_prompt_by_hash(generate_prompt_hash("Lonely sunset"))
+        self.assertIsNotNone(again)
+        self.assertNotEqual(again["id"], plain)
+        self.assertEqual(self.archived(), [])

tests/test_prompt_archive_schema.py

@@ -0,0 +1,196 @@
+"""#443: the Library parent archive table, its migration and its validation."""
+
+import shutil
+import sqlite3
+import tempfile
+import unittest
+from pathlib import Path
+
+from database.models import PromptModel
+from database.operations import PromptDatabase
+from database.prompt_archive import (
+    TABLE,
+    migrate_prompt_archive,
+    prompt_archive_schema_error,
+)
+from utils.hashing import generate_prompt_hash
+
+
+def _tables(conn):
+    return {
+        row[0]
+        for row in conn.execute("SELECT name FROM sqlite_master WHERE type='table'")
+    }
+
+
+class TestPromptArchiveSchema(unittest.TestCase):
+    def setUp(self):
+        self.temp = tempfile.TemporaryDirectory()
+        self.path = str(Path(self.temp.name) / "archive.db")
+        self.db = PromptDatabase(self.path)
+
+    def tearDown(self):
+        self.db.close_all()
+        self.temp.cleanup()
+
+    def conn(self):
+        return self.db.model.get_connection()
+
+    def save(self, text):
+        return self.db.save_prompt(text, prompt_hash=generate_prompt_hash(text))
+
+    def test_startup_creates_the_table_and_it_validates(self):
+        self.assertIn(TABLE, _tables(self.conn()))
+        self.assertIsNone(prompt_archive_schema_error(self.conn(), allow_missing=False))
+
+    def test_migration_is_idempotent(self):
+        conn = self.conn()
+        migrate_prompt_archive(conn)
+        migrate_prompt_archive(conn)
+        self.assertIsNone(prompt_archive_schema_error(conn, allow_missing=False))
+
+    def test_a_missing_table_is_accepted_only_when_allowed(self):
+        conn = sqlite3.connect(":memory:")
+        self.addCleanup(conn.close)
+        self.assertIsNone(prompt_archive_schema_error(conn))
+        self.assertEqual(
+            prompt_archive_schema_error(conn, allow_missing=False),
+            "Missing prompt archive schema",
+        )
+
+    def test_deleting_a_prompt_drops_its_archive_row(self):
+        pid = self.save("Archived then deleted")
+        conn = self.conn()
+        conn.execute(
+            f"INSERT INTO {TABLE}(prompt_id, archived_at) VALUES (?, ?)",
+            (pid, "2026-10-28T00:00:00+00:00"),
+        )
+        conn.commit()
+        self.db.delete_prompt(pid)
+        self.assertEqual(conn.execute(f"SELECT COUNT(*) FROM {TABLE}").fetchone()[0], 0)
+
+    def _broken(self, *statements):
+        conn = sqlite3.connect(":memory:")
+        self.addCleanup(conn.close)
+        conn.execute("CREATE TABLE prompts (id INTEGER PRIMARY KEY)")
+        for statement in statements:
+            conn.execute(statement)
+        return conn
+
+    def test_unsupported_structures_are_refused(self):
+        good = (
+            f"CREATE TABLE {TABLE} (prompt_id INTEGER PRIMARY KEY, "
+            "archived_at TEXT NOT NULL, "
+            "FOREIGN KEY(prompt_id) REFERENCES prompts(id) ON DELETE CASCADE)"
+        )
+        cases = {
+            "other definition": (
+                f"CREATE TABLE {TABLE} (prompt_id INTEGER PRIMARY KEY, "
+                "archived_at TEXT)",
+            ),
+            "no cascade": (
+                f"CREATE TABLE {TABLE} (prompt_id INTEGER PRIMARY KEY, "
+                "archived_at TEXT NOT NULL, "
+                "FOREIGN KEY(prompt_id) REFERENCES prompts(id))",
+            ),
+            "extra index": (good, f"CREATE INDEX x ON {TABLE}(archived_at)"),
+            "trigger": (
+                good,
+                f"CREATE TRIGGER t AFTER INSERT ON {TABLE} BEGIN SELECT 1; END",
+            ),
+            "view name": (f"CREATE VIEW {TABLE} AS SELECT 1",),
+        }
+        for name, statements in cases.items():
+            with self.subTest(name):
+                self.assertIsNotNone(
+                    prompt_archive_schema_error(self._broken(*statements))
+                )
+
+    def test_invalid_values_and_dangling_rows_are_refused(self):
+        good = (
+            f"CREATE TABLE {TABLE} (prompt_id INTEGER PRIMARY KEY, "
+            "archived_at TEXT NOT NULL, "
+            "FOREIGN KEY(prompt_id) REFERENCES prompts(id) ON DELETE CASCADE)"
+        )
+        for name, row in {
+            "number": "(1, 5)",
+            "blob": "(1, x'00')",
+            "long": f"(1, '{'x' * 65}')",
+            "dangling": "(99, '2026-10-28T00:00:00+00:00')",
+        }.items():
+            with self.subTest(name):
+                conn = self._broken(good, "INSERT INTO prompts(id) VALUES (1)")
+                conn.execute(f"INSERT INTO {TABLE} VALUES {row}")
+                self.assertIsNotNone(prompt_archive_schema_error(conn))
+
+    def test_a_refused_structure_stops_the_migration_without_changes(self):
+        conn = self._broken(
+            f"CREATE TABLE {TABLE} (prompt_id INTEGER PRIMARY KEY, archived_at TEXT)"
+        )
+        with self.assertRaises(sqlite3.DatabaseError):
+            migrate_prompt_archive(conn)
+        sql = conn.execute(
+            "SELECT sql FROM sqlite_master WHERE name=?", (TABLE,)
+        ).fetchone()[0]
+        self.assertNotIn("NOT NULL", sql)
+
+
+class TestPromptArchiveBackups(unittest.TestCase):
+    """Restore validation accepts an older backup and refuses a damaged table."""
+
+    def setUp(self):
+        self.temp = tempfile.TemporaryDirectory()
+        self.root = Path(self.temp.name)
+        self.db = PromptDatabase(str(self.root / "live.db"))
+
+    def tearDown(self):
+        self.db.close_all()
+        self.temp.cleanup()
+
+    def copy(self, name):
+        target = self.root / name
+        self.db.model.backup_database(str(target))
+        return str(target)
+
+    def test_a_backup_without_the_table_verifies_and_migrates(self):
+        path = self.copy("old.db")
+        with sqlite3.connect(path) as conn:
+            conn.execute(f"DROP TABLE {TABLE}")
+        conn.close()
+        ok, reason = self.db.model.verify_database_file(path)
+        self.assertTrue(ok, reason)
+        model = PromptModel(path)
+        try:
+            self.assertIn(TABLE, _tables(model.get_connection()))
+        finally:
+            model.close_all()
+
+    def test_a_backup_with_a_damaged_table_is_refused(self):
+        path = self.copy("bad.db")
+        with sqlite3.connect(path) as conn:
+            conn.execute(f"DROP TABLE {TABLE}")
+            conn.execute(f"CREATE TABLE {TABLE} (prompt_id INTEGER, archived_at TEXT)")
+        conn.close()
+        ok, reason = self.db.model.verify_database_file(path)
+        self.assertFalse(ok)
+        self.assertIn("archive", reason.lower())
+
+    def test_an_archived_prompt_survives_a_backup_copy(self):
+        pid = self.db.save_prompt("Kept", prompt_hash=generate_prompt_hash("Kept"))
+        conn = self.db.model.get_connection()
+        conn.execute(
+            f"INSERT INTO {TABLE}(prompt_id, archived_at) VALUES (?, ?)",
+            (pid, "2026-10-28T00:00:00+00:00"),
+        )
+        conn.commit()
+        path = self.copy("copy.db")
+        shutil.copy(path, path + ".2")
+        with sqlite3.connect(path + ".2") as other:
+            self.assertEqual(
+                other.execute(f"SELECT prompt_id FROM {TABLE}").fetchall(), [(pid,)]
+            )
+        other.close()
+
+
+if __name__ == "__main__":
+    unittest.main()