diff --git a/TESTING.md b/TESTING.md
index 5dff959d..5d7db36d 100644
--- a/TESTING.md
+++ b/TESTING.md
@@ -2499,6 +2499,7 @@ node --test tests/js/workspace-system-state.test.js tests/js/workspace-system-co
   tests/js/workspace-inspection-state.test.js tests/js/workspace-images-state.test.js \
   tests/js/workspace-system-sources.test.js tests/js/workspace-thumbnail-review.test.js
 python3 -m unittest tests.test_settings_sources tests.test_scan_scope -v
+python3 -m unittest tests.test_maintenance_plans tests.test_api_maintenance -v
 npm run --prefix tools/vendor test:workflows -- system-concept.spec.mjs workspace-system.spec.mjs \
   workspace-system-sources.spec.mjs \
   --project chromium --project touch --project firefox --project webkit
@@ -2531,6 +2532,44 @@ answers `storage-changed`. Errors are coded (`invalid-request`, `too-large`,
 `storage-changed`, or a null code with "Failed to review maintenance") and carry no
 path or exception text.
 
+`POST /prompt_manager/maintenance/apply {token}` starts the reviewed plan as an
+Activity job (202 `{job}`): kind `maintenance`, titled after the concept's row
+("Duplicate prompt text", "Optimize database", "Unused prompt components", "Missing-file
+links"), admission `records` (it never runs beside a scan, restore, LoRA import or other
+record-changing job, legacy-page holds included: `test_apply_refused_while_restore_runs`),
+not cancellable, with the groups `changed`, `protected` and `errors`. The token is consumed
+on the loop turn the job starts, so a second apply answers 409 `plan-used` with that job;
+a refused start (`busy`) keeps it. Other errors: `invalid-request`, `plan-expired` (410:
+unknown, expired, or forgotten after a restart) and `storage-changed` (409: a restore
+replaced the database since the review; the plan is dropped). The job:
+
+1. takes a safety backup when the plan was reviewed with the policy on or the policy is
+   on now (`maintenance.safety_backup` in `config.json`, default `true`; GET/POST
+   `/prompt_manager/settings` field `safety_backup` with the revision). The copy is made
+   with SQLite's backup API into a temp file beside the database, switched to a
+   rollback journal (so a later read-only open leaves no `-wal`/`-shm` beside it),
+   verified with `verify_database_file`, fsynced, made `0o600` on POSIX and only then
+   renamed with `os.replace` onto `<db>.backup_<YYYYmmdd_HHMMSS_ffffff>`, reserved with
+   `O_EXCL` (`_1`, `_2`… when a coarse clock repeats a stamp:
+   `test_backup_names_unique_within_one_second`). A backup file on disk is therefore
+   always a verified one. Any failure removes what it wrote and ends the job
+   `backup-failed` before anything is changed (`test_backup_failure_aborts_before_write`);
+2. applies the plan: inside `BEGIN IMMEDIATE` the fingerprint is computed again on the
+   write connection and a difference ends the job `plan-changed` with nothing written
+   (`test_apply_rejects_changed_target`; a missing link whose file is back or whose source
+   went away is never removed). The targets are deleted in batches of 400 ids in that one
+   transaction. Missing-file links remove database rows only; every media file's bytes
+   are unchanged (`test_missing_links_never_deletes_files`). VACUUM runs outside a
+   transaction and binds to its database only;
+3. counts the changed and protected records (outcome rows hold root-relative paths or
+   prompt ids only), logs the backup by its file name, and prunes safety backups to the
+   newest 10. With no backup the log and the status line say "No safety backup".
+
+A failure the job cannot name ends it "Maintenance failed; nothing was changed." with a
+null code and no exception text. Windows: every connection the backup and the previews
+open is closed before the temp copy is renamed or removed, and the suites' teardown
+asserts no handle under the temp root is still open.
+
 The legacy `POST /prompt_manager/maintenance` keeps its shape. Its `vacuum` result is
 `success:false` with the error when VACUUM fails (`vacuum_database` now raises after
 logging), and `cleanup_orphaned_images` removes only links under a source that can be
diff --git a/database/maintenance.py b/database/maintenance.py
index 0abcd16d..1ebe789e 100644
--- a/database/maintenance.py
+++ b/database/maintenance.py
@@ -50,6 +50,7 @@ MAINTENANCE_OPERATIONS = (
     "missing_links",
 )
 MAX_DETAILS = 50  # detail rows a plan lists
+MAX_RESULT_ROWS = 200  # changed records an apply describes (the job lists 200)
 EXCERPT_CHARS = 80
 # Ids bound per statement: the pair-reference read binds each id twice, which
 # stays under SQLite's historical 999-variable limit.
@@ -220,9 +221,9 @@ def _add_prompt_state(fingerprint, conn, ids):
                 fingerprint.add(list(row))
 
 
-def _texts(conn, ids):
-    """``{id: text}`` of at most MAX_DETAILS prompts, for detail rows."""
-    ids = list(ids)[:MAX_DETAILS]
+def _texts(conn, ids, limit=MAX_DETAILS):
+    """``{id: text}`` of the first ``limit`` prompts, for detail rows."""
+    ids = list(ids)[:limit]
     if not ids:
         return {}
     rows = conn.execute(
@@ -548,3 +549,81 @@ class MaintenanceOperations:
                 },
             ],
         )
+
+    # ── apply ─────────────────────────────────────────────────────────
+
+    def apply_maintenance(self, operation, fingerprint, targets, roots=()) -> dict:
+        """Apply a reviewed plan exactly, or raise PlanChanged (blocking).
+
+        Inside ``BEGIN IMMEDIATE`` the fingerprint is computed again on the
+        write connection; when it differs nothing is written. ``targets`` and
+        ``roots`` are the plan's. Returns ``{changed, rows}``: how many
+        records changed and up to ``MAX_RESULT_ROWS`` descriptions of them
+        (``{prompt_id[, kept_id]}`` or ``{image_path}``).
+        """
+        if operation not in MAINTENANCE_OPERATIONS:
+            raise ValueError(f"Unknown maintenance operation {operation!r}")
+        if operation == "optimize":
+            # VACUUM cannot run in a transaction; the plan binds the database.
+            if fingerprint != _Fingerprint("optimize").hexdigest():
+                raise PlanChanged("optimize")
+            self.model.vacuum_database()
+            return {"changed": 1, "rows": []}
+        appliers = {
+            "duplicate_prompts": self._apply_duplicate_prompts,
+            "unused_components": self._apply_unused_components,
+            "missing_links": self._apply_missing_links,
+        }
+        with self._pair_transaction() as conn:
+            return appliers[operation](conn, fingerprint, targets, roots)
+
+    def _apply_duplicate_prompts(self, conn, fingerprint, targets, _roots):
+        groups, _paired, current = self._duplicates_fingerprint(conn)
+        if current != fingerprint or groups != targets:
+            raise PlanChanged("duplicate_prompts")
+        changed, rows = 0, []
+        for sorted_ids in groups:
+            removed, _moved = self._merge_duplicate_group(conn, sorted_ids)
+            changed += removed
+            rows.extend(
+                {"prompt_id": duplicate_id, "kept_id": sorted_ids[0]}
+                for duplicate_id in sorted_ids[1:]
+            )
+        return {"changed": changed, "rows": rows[:MAX_RESULT_ROWS]}
+
+    def _apply_unused_components(self, conn, fingerprint, targets, _roots):
+        ids, current = self._unused_fingerprint(conn)
+        if current != fingerprint or ids != targets:
+            raise PlanChanged("unused_components")
+        texts = _texts(conn, ids, limit=MAX_RESULT_ROWS)
+        changed = _delete_ids(conn, "prompts", ids)
+        rows = [
+            {"prompt_id": prompt_id, "text": _excerpt(texts.get(prompt_id))}
+            for prompt_id in ids[:MAX_RESULT_ROWS]
+        ]
+        return {"changed": changed, "rows": rows}
+
+    def _apply_missing_links(self, conn, fingerprint, targets, roots):
+        rows = []
+        for batch in _batches(sorted(targets)):
+            rows.extend(
+                tuple(row)
+                for row in conn.execute(
+                    "SELECT id, image_path, filename, file_path FROM generated_images "
+                    f"WHERE id IN ({_marks(batch)}) ORDER BY id",  # nosec B608
+                    batch,
+                )
+            )
+        readable = readable_roots(roots)
+        candidates = []
+        for row, owner, state in link_states(rows, roots, readable):
+            if state != "missing":  # back, or its source went away
+                raise PlanChanged("missing_links")
+            candidates.append((row, owner))
+        if self._links_fingerprint(candidates) != fingerprint:
+            raise PlanChanged("missing_links")
+        changed = _delete_ids(conn, "generated_images", [row[0] for row in rows])
+        return {
+            "changed": changed,
+            "rows": [{"image_path": row[1]} for row in rows[:MAX_RESULT_ROWS]],
+        }
diff --git a/py/api/maintenance.py b/py/api/maintenance.py
index ca2df99b..7fce44c4 100644
--- a/py/api/maintenance.py
+++ b/py/api/maintenance.py
@@ -16,19 +16,49 @@ code, error}`` with ``invalid-request`` (400), ``too-large`` (413: the body,
 or a plan over the store's budget), ``storage-changed`` (409: the database
 was replaced while the preview read it) or a null code (500); none carries a
 server path or exception text.
+
+``POST /prompt_manager/maintenance/apply {token}`` applies a reviewed plan as
+an Activity job (202 ``{job}``): kind ``maintenance``, titled after the
+operation, admission ``records`` (refused while a scan, restore, LoRA import
+or other record-changing job runs, and refusing them meanwhile), not
+cancellable, with outcome groups ``changed``, ``protected`` and ``errors``.
+The token is consumed when the job starts (on the same loop turn, before its
+runner runs); a refused start keeps it. The runner:
+
+1. takes a safety backup when the plan was reviewed with the policy on or the
+   policy (``maintenance.safety_backup``, default on) is on now: SQLite's
+   backup API into a temp file beside the database, ``verify_database_file``,
+   fsync, ``0o600`` on POSIX, then ``os.replace`` onto a unique
+   ``<db>.backup_<YYYYmmdd_HHMMSS_ffffff>[_n]`` name. Any failure removes
+   what it wrote and ends the job ``backup-failed`` with nothing changed;
+2. applies the plan, which first rechecks the reviewed fingerprint inside
+   ``BEGIN IMMEDIATE`` (``plan-changed`` writes nothing);
+3. counts the changed and protected records, and prunes old safety backups.
+
+The job log names the backup by its file name only, and the job says "No
+safety backup" when none was taken. Apply errors: ``invalid-request`` (400),
+``plan-expired`` (410: unknown, expired or forgotten after a restart),
+``plan-used`` (409, with the job that applied it), ``storage-changed`` (409:
+a restore replaced the database since the review) and ``busy`` (409).
 """
 
 import datetime
 import os
+import sqlite3
+import tempfile
+from contextlib import closing
 
 from aiohttp import web
 
-from .maintenance_plans import ScopeTooLarge
+from .admin import _prune_safety_backups, _scan_location, _scan_roots
+from .job_routes import busy_response
+from .jobs import MAINTENANCE_JOB_KIND, JobBusy
+from .maintenance_plans import PlanExpired, PlanUsed, ScopeTooLarge
 
 try:
-    from ...database.maintenance import MAINTENANCE_OPERATIONS
+    from ...database.maintenance import MAINTENANCE_OPERATIONS, PlanChanged
 except ImportError:
-    from database.maintenance import MAINTENANCE_OPERATIONS
+    from database.maintenance import MAINTENANCE_OPERATIONS, PlanChanged
 
 PREVIEW_BODY_MAX = 4096
 # The concept's Maintenance row titles, which name each operation's job.
@@ -39,6 +69,117 @@ MAINTENANCE_TITLES = {
     "missing_links": "Missing-file links",
 }
 _OPERATION_ERROR = "operation must be one of: " + ", ".join(MAINTENANCE_OPERATIONS)
+MAINTENANCE_GROUPS = ("changed", "protected", "errors")
+MAX_TOKEN_LENGTH = 128
+BACKUP_FAILED = (
+    "The safety backup could not be taken and verified; nothing was changed."
+)
+PLAN_CHANGED = (
+    "The library changed since the review; nothing was changed. "
+    "Review the plan again."
+)
+APPLY_FAILED = "Maintenance failed; nothing was changed."
+NO_BACKUP = "No safety backup"
+_NAME_ATTEMPTS = 1000
+
+
+class SafetyBackupFailed(Exception):
+    """The safety backup could not be written, verified or put in place."""
+
+
+def _now():
+    return datetime.datetime.now()
+
+
+def _changed_line(operation, count):
+    if operation == "optimize":
+        return "Optimized the database"
+    if operation == "duplicate_prompts":
+        noun = "duplicate prompt" if count == 1 else "duplicate prompts"
+        return f"Merged {count:,} {noun}"
+    if operation == "unused_components":
+        noun = "unused prompt component" if count == 1 else "unused prompt components"
+        return f"Removed {count:,} {noun}"
+    noun = "link to a missing file" if count == 1 else "links to missing files"
+    return f"Removed {count:,} {noun}"
+
+
+def _remove_quietly(*paths):
+    for path in paths:
+        try:
+            if path and os.path.exists(path):
+                os.remove(path)
+        except OSError:
+            pass
+
+
+def _reserve_backup_name(db_path):
+    """Create and return an unused ``<db>.backup_<stamp>[_n]`` (O_EXCL)."""
+    base = f"{db_path}.backup_{_now().strftime('%Y%m%d_%H%M%S_%f')}"
+    for attempt in range(_NAME_ATTEMPTS):
+        candidate = base if attempt == 0 else f"{base}_{attempt}"
+        try:
+            fd = os.open(candidate, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600)
+        except FileExistsError:
+            continue
+        os.close(fd)
+        return candidate
+    raise SafetyBackupFailed("No free backup name")
+
+
+def _standalone(path):
+    """Make the copy one self-contained file (rollback journal, not WAL).
+
+    The backup API copies the live database's WAL flag; a WAL file opened
+    read-only later (a verify, a restore preview) would leave ``-wal`` and
+    ``-shm`` sidecars beside it.
+    """
+    try:
+        with closing(sqlite3.connect(path)) as conn:
+            conn.execute("PRAGMA journal_mode=DELETE").fetchone()
+    except sqlite3.Error as exc:
+        raise SafetyBackupFailed("The backup could not be finished") from exc
+
+
+def _fsync(path):
+    with open(path, "rb+") as handle:
+        os.fsync(handle.fileno())
+
+
+def take_safety_backup(model):
+    """Write and verify a safety backup; return its file name (blocking).
+
+    The copy is made in a temp file beside the database and only renamed
+    onto its final, unique name once it verified, so a backup file on disk
+    is always a verified one. Raises SafetyBackupFailed, leaving no file.
+    """
+    db_path = os.path.abspath(model.db_path)
+    fd, temp = tempfile.mkstemp(
+        dir=os.path.dirname(db_path), prefix=".pm-safety-", suffix=".tmp"
+    )
+    os.close(fd)
+    final = None
+    try:
+        if not model.backup_database(temp):
+            raise SafetyBackupFailed("The backup could not be written")
+        _standalone(temp)
+        ok, _reason = model.verify_database_file(temp)
+        if not ok:
+            raise SafetyBackupFailed("The backup did not verify")
+        _fsync(temp)
+        if os.name != "nt":
+            os.chmod(temp, 0o600)
+        final = _reserve_backup_name(db_path)
+        os.replace(temp, final)
+        return os.path.basename(final)
+    except OSError as exc:
+        _remove_quietly(final)
+        raise SafetyBackupFailed("The backup could not be put in place") from exc
+    except SafetyBackupFailed:
+        _remove_quietly(final)
+        raise
+    finally:
+        _remove_quietly(temp, temp + "-journal", temp + "-wal", temp + "-shm")
 
 
 def _failure(message, status, code=None, **extra):
@@ -63,6 +204,10 @@ class MaintenanceRoutesMixin:
         async def maintenance_preview_route(request):
             return await self.preview_maintenance(request)
 
+        @routes.post("/prompt_manager/maintenance/apply")
+        async def maintenance_apply_route(request):
+            return await self.apply_maintenance_plan(request)
+
     def _database_identity(self, db):
         """What binds a plan to one database: the swap generation and its file."""
         path = os.path.normcase(os.path.realpath(db.model.db_path))
@@ -71,7 +216,9 @@ class MaintenanceRoutesMixin:
     @staticmethod
     def _safety_backup_policy():
         """``on`` or ``off``: whether an apply takes a safety backup first."""
-        return "on"
+        from ..config import PromptManagerConfig
+
+        return "on" if PromptManagerConfig.SAFETY_BACKUP else "off"
 
     async def _read_maintenance_body(self, request, allowed):
         """``(data, None)`` for a capped JSON object with only ``allowed`` keys."""
@@ -150,3 +297,138 @@ class MaintenanceRoutesMixin:
             },
             headers={"Cache-Control": "no-store"},
         )
+
+    # ── apply ─────────────────────────────────────────────────────────
+
+    async def apply_maintenance_plan(self, request):
+        data, error = await self._read_maintenance_body(request, ("token",))
+        if error is not None:
+            return error
+        token = data.get("token")
+        if not isinstance(token, str) or not 0 < len(token) <= MAX_TOKEN_LENGTH:
+            return _invalid("token must be the plan's token")
+        store = self._maintenance_plans
+        try:
+            plan = store.peek(token)
+        except PlanUsed as used:
+            job = self._jobs.get(used.job_id)
+            return _failure(
+                "This plan was already applied.",
+                409,
+                "plan-used",
+                job={"id": used.job_id, "title": job.title if job else None},
+            )
+        except PlanExpired:
+            return _failure(
+                "This review expired. Review the plan again.", 410, "plan-expired"
+            )
+        db = self.db
+        if plan.db_identity != self._database_identity(db):
+            store.discard(token)
+            return _failure(
+                "The database was replaced since the review. Review again.",
+                409,
+                "storage-changed",
+            )
+        try:
+            job = self._jobs.start(
+                MAINTENANCE_JOB_KIND,
+                MAINTENANCE_TITLES[plan.operation],
+                admission="records",
+                cancellable=False,
+                scope={"operation": plan.operation},
+                groups=MAINTENANCE_GROUPS,
+                runner=lambda job: self._run_maintenance_plan(job, db, plan),
+            )
+        except JobBusy as busy:
+            return busy_response(busy)  # the token stays usable
+        # Same loop turn as the start: the runner has not run yet.
+        store.consume(token, plan.operation, job_id=job.id)
+        return web.json_response({"success": True, "job": job.summary()}, status=202)
+
+    async def _run_maintenance_plan(self, job, db, plan):
+        payload = plan.payload
+        backup = None
+        if payload["backup"] == "on" or self._safety_backup_policy() == "on":
+            job.update(status_line="Taking a safety backup")
+            job.log_line("Taking a safety backup of the database")
+            try:
+                backup = await self._run_in_executor(take_safety_backup, db.model)
+            except SafetyBackupFailed:
+                self.logger.error("Safety backup failed", exc_info=True)
+                return _fail_job(job, "backup-failed", BACKUP_FAILED)
+            job.log_line(f"Safety backup {backup} verified")
+        else:
+            job.log_line("No safety backup (the policy is off)")
+        job.update(status_line=f"Applying: {job.title}")
+        try:
+            result = await self._run_in_executor(
+                db.apply_maintenance,
+                plan.operation,
+                plan.fingerprint,
+                payload["targets"],
+                payload["roots"],
+            )
+        except PlanChanged:
+            return _fail_job(job, "plan-changed", PLAN_CHANGED)
+        except Exception:
+            self.logger.exception(f"Maintenance {plan.operation} failed")
+            return _fail_job(job, None, APPLY_FAILED)
+        rows = await self._run_in_executor(
+            self._maintenance_rows, plan.operation, result["rows"]
+        )
+        job.record_counted("changed", result["changed"], rows)
+        job.record_counted("protected", payload["protected"])
+        line = _changed_line(plan.operation, result["changed"])
+        job.log_line(line)
+        if backup:
+            removed = await self._run_in_executor(
+                _prune_safety_backups, db.model.db_path
+            )
+            if removed:
+                job.log_line(f"Removed {len(removed)} older safety backups")
+        job.update(
+            progress=100,
+            status_line=f"{line} · "
+            + (f"Safety backup {backup}" if backup else NO_BACKUP),
+        )
+
+    def _maintenance_rows(self, operation, described):
+        """Outcome rows for the changed records (root-relative paths only)."""
+        if operation == "missing_links":
+            roots = _scan_roots(self._get_all_output_dirs())
+            rows = []
+            for item in described:
+                path, label, index = _scan_location(item["image_path"], roots)
+                rows.append(
+                    {
+                        "path": path,
+                        "root": label,
+                        "root_index": index,
+                        "detail": "Link removed; the file was already missing",
+                    }
+                )
+            return rows
+        if operation == "duplicate_prompts":
+            return [
+                {
+                    "path": "",
+                    "detail": (
+                        f"Prompt {item['prompt_id']} merged into "
+                        f"prompt {item['kept_id']}"
+                    ),
+                }
+                for item in described
+            ]
+        return [
+            {"path": "", "detail": f"Removed: {item.get('text') or item['prompt_id']}"}
+            for item in described
+        ]
+
+
+def _fail_job(job, code, message):
+    """End ``job`` failed with ``message`` (one error row) and its code."""
+    job.record("errors", {"path": "", "detail": message}, f"! {message}")
+    job.update(status_line=message)
+    job.publish({"type": "error", "code": code, "message": message})
+    job.finish("failed")
diff --git a/py/api/settings_routes.py b/py/api/settings_routes.py
index 4760e083..17637f7e 100644
--- a/py/api/settings_routes.py
+++ b/py/api/settings_routes.py
@@ -40,6 +40,7 @@ _SETTINGS_FILE_KEYS = {
     "INFINITE_SCROLL": ("web_ui", "infinite_scroll"),
     "WORKER_THREADS": ("performance", "worker_threads"),
     "MONITORING_ENABLED": ("gallery", "monitoring", "enabled"),
+    "SAFETY_BACKUP": ("maintenance", "safety_backup"),
 }
 # Attributes that live on GalleryConfig; the others on PromptManagerConfig.
 _GALLERY_SETTINGS = frozenset({"MONITORING_ENABLED"})
@@ -187,6 +188,7 @@ class SettingsRoutesMixin:
                         "webui_display_mode": PromptManagerConfig.WEBUI_DISPLAY_MODE,
                         "infinite_scroll": PromptManagerConfig.INFINITE_SCROLL,
                         "worker_threads": PromptManagerConfig.WORKER_THREADS,
+                        "safety_backup": PromptManagerConfig.SAFETY_BACKUP,
                         "cpu_count": PromptManagerConfig.max_worker_threads(),
                         "gallery_root_paths": root_paths,
                         "gallery_root_path": root_paths[0] if root_paths else "",
@@ -481,6 +483,10 @@ class SettingsRoutesMixin:
             if not isinstance(data["monitoring_enabled"], bool):
                 raise ValueError("monitoring_enabled must be true or false")
             updates["MONITORING_ENABLED"] = data["monitoring_enabled"]
+        if "safety_backup" in data:
+            if not isinstance(data["safety_backup"], bool):
+                raise ValueError("safety_backup must be true or false")
+            updates["SAFETY_BACKUP"] = data["safety_backup"]
         if "worker_threads" in data:
             updates["WORKER_THREADS"] = validate_worker_threads(
                 data["worker_threads"], PromptManagerConfig.max_worker_threads()
diff --git a/py/config.py b/py/config.py
index c6fa7197..18f94f10 100644
--- a/py/config.py
+++ b/py/config.py
@@ -659,6 +659,10 @@ class PromptManagerConfig:
     DEFAULT_WORKER_THREADS = max(1, (os.cpu_count() or 1) // 2)
     WORKER_THREADS = DEFAULT_WORKER_THREADS
 
+    # Maintenance: a reviewed apply takes a verified safety backup of the
+    # database first (System › Backup & export › Recovery policy, #249 S6).
+    SAFETY_BACKUP = True
+
     @classmethod
     def max_worker_threads(cls) -> int:
         """Largest accepted worker-thread count: the detected core count."""
@@ -713,6 +717,7 @@ class PromptManagerConfig:
                 "auto_backup_interval": cls.AUTO_BACKUP_INTERVAL,
                 "worker_threads": cls.WORKER_THREADS,
             },
+            "maintenance": {"safety_backup": cls.SAFETY_BACKUP},
             "gallery": GalleryConfig.get_config(),
             "integrations": IntegrationConfig.get_config(),
             "autotag": AutotagConfig.get_config(),
@@ -848,6 +853,12 @@ class PromptManagerConfig:
                 performance["worker_threads"], cls.WORKER_THREADS
             )
 
+        maintenance = new_config.get("maintenance", {})
+        if isinstance(maintenance, dict) and isinstance(
+            maintenance.get("safety_backup"), bool
+        ):
+            cls.SAFETY_BACKUP = maintenance["safety_backup"]
+
         # Update gallery config
         if "gallery" in new_config:
             GalleryConfig.update_config(new_config["gallery"])
diff --git a/tests/test_api_maintenance.py b/tests/test_api_maintenance.py
index 50adf750..f03581c1 100644
--- a/tests/test_api_maintenance.py
+++ b/tests/test_api_maintenance.py
@@ -6,8 +6,11 @@ exactly that plan. Errors are coded and never carry a server path or an
 exception text.
 """
 
+import asyncio
+import datetime
 import hashlib
 import json
+import re
 import os
 import sqlite3
 import sys
@@ -229,6 +232,440 @@ class TestPreviewRoute(MaintenanceAPITestCase):
             restarted.db.close_all()
 
 
+APPLY = "/prompt_manager/maintenance/apply"
+BACKUP_NAME = re.compile(r"^prompts\.db\.backup_\d{8}_\d{6}_\d{6}(_\d+)?$")
+
+
+class ApplyTestCase(MaintenanceAPITestCase):
+    def backups(self):
+        return sorted(
+            name
+            for name in os.listdir(self.tmpdir)
+            if name.startswith("prompts.db.backup_")
+        )
+
+    def strays(self):
+        """Files beside the database that are neither it nor a safety backup."""
+        known = {"prompts.db", "prompts.db-wal", "prompts.db-shm", "config.json"}
+        return sorted(
+            name
+            for name in os.listdir(self.tmpdir)
+            if os.path.isfile(os.path.join(self.tmpdir, name))
+            and name not in known
+            and not BACKUP_NAME.match(name)
+        )
+
+    async def apply(self, token, status=202):
+        return await self.post(APPLY, {"token": token}, status)
+
+    async def run_apply(self, operation):
+        plan = await self.preview(operation)
+        body = await self.apply(plan["token"])
+        job = self.api._jobs.get(body["job"]["id"])
+        await asyncio.wait_for(asyncio.shield(job.task), WAIT)
+        return plan, body["job"], job
+
+
+class TestApplyRoute(ApplyTestCase):
+    async def test_apply_runs_as_a_records_job_with_a_verified_backup(self):
+        orphan = self.save("Orphan")
+        plan, summary, job = await self.run_apply("unused_components")
+        self.assertEqual(summary["kind"], "maintenance")
+        self.assertEqual(summary["title"], "Unused prompt components")
+        self.assertEqual(summary["admission"], "records")
+        self.assertFalse(summary["cancellable"])
+        self.assertEqual(
+            summary["counters"], {"changed": 0, "protected": 0, "errors": 0}
+        )
+        self.assertEqual(summary["scope"], {"operation": "unused_components"})
+        self.assertEqual(job.state, "finished", job.log)
+        self.assertEqual(job.counters, {"changed": 1, "protected": 0, "errors": 0})
+        self.assertIsNone(self.api.db.get_prompt_by_id(orphan))
+        [backup] = self.backups()
+        self.assertRegex(backup, BACKUP_NAME)
+        ok, _reason = self.api.db.model.verify_database_file(
+            os.path.join(self.tmpdir, backup)
+        )
+        self.assertTrue(ok)
+        self.assertEqual(self.strays(), [])
+        self.assertIn(f"Safety backup {backup} verified", "\n".join(job.log))
+        self.assertEqual(
+            job.status_line,
+            f"Removed 1 unused prompt component · Safety backup {backup}",
+        )
+        self.assertEqual(job.outcomes["changed"][0]["detail"], "Removed: Orphan")
+        report = json.dumps(job.report())
+        self.assertNotIn(self.tmpdir, report)
+        if os.name == "nt":  # pragma: no cover - native path spelling
+            self.assertNotIn(self.tmpdir.replace("\\", "/"), report)
+
+    async def test_apply_rejects_changed_target(self):
+        orphan, soon = self.save("Orphan"), self.save("Soon a recipe")
+        plan = await self.preview("unused_components")
+        with self.api.db.model.get_connection() as conn:
+            conn.execute(
+                "INSERT INTO prompt_library_state(prompt_id,title) VALUES(?,?)",
+                (soon, "Recipe"),
+            )
+        body = await self.apply(plan["token"])
+        job = self.api._jobs.get(body["job"]["id"])
+        await asyncio.wait_for(asyncio.shield(job.task), WAIT)
+        self.assertEqual(job.state, "failed")
+        self.assertEqual(job.last_event["code"], "plan-changed")
+        self.assertEqual(
+            job.status_line,
+            "The library changed since the review; nothing was changed. "
+            "Review the plan again.",
+        )
+        self.assertEqual(job.counters["errors"], 1)
+        self.assertIsNotNone(self.api.db.get_prompt_by_id(orphan))
+        self.assertIsNotNone(self.api.db.get_prompt_by_id(soon))
+
+    async def test_token_is_single_use(self):
+        self.save("Orphan")
+        plan, summary, job = await self.run_apply("unused_components")
+        again = await self.apply(plan["token"], 409)
+        self.assertEqual(again["code"], "plan-used")
+        self.assertEqual(
+            again["job"], {"id": summary["id"], "title": "Unused prompt components"}
+        )
+        kinds = [j.kind for j in self.api._jobs.list()]
+        self.assertEqual(kinds.count("maintenance"), 1)
+
+    async def test_apply_refused_while_restore_runs(self):
+        orphan = self.save("Orphan")
+        plan = await self.preview("unused_components")
+        async with self.api._jobs.hold(
+            "restore", "Database restore (admin page)", admission="records"
+        ) as restore:
+            busy = await self.apply(plan["token"], 409)
+            self.assertEqual(busy["code"], "busy")
+            self.assertEqual(
+                busy["running"],
+                {"id": restore.id, "title": "Database restore (admin page)"},
+            )
+        self.assertIsNotNone(self.api.db.get_prompt_by_id(orphan))
+        # The refused apply kept the token: it applies once the restore ended.
+        body = await self.apply(plan["token"])
+        job = self.api._jobs.get(body["job"]["id"])
+        await asyncio.wait_for(asyncio.shield(job.task), WAIT)
+        self.assertIsNone(self.api.db.get_prompt_by_id(orphan))
+
+    async def test_backup_failure_aborts_before_write(self):
+        orphan = self.save("Orphan")
+        model = self.api.db.model
+        failures = {
+            "backup": patch.object(model, "backup_database", return_value=False),
+            "verify": patch.object(
+                model, "verify_database_file", return_value=(False, "broken")
+            ),
+        }
+        for name, failure in failures.items():
+            with self.subTest(failure=name):
+                with failure, patch.object(self.api.db, "apply_maintenance") as applied:
+                    _plan, _summary, job = await self.run_apply("unused_components")
+                applied.assert_not_called()
+                self.assertEqual(job.state, "failed")
+                self.assertEqual(job.last_event["code"], "backup-failed")
+                self.assertEqual(
+                    job.status_line,
+                    "The safety backup could not be taken and verified; "
+                    "nothing was changed.",
+                )
+                self.assertIsNotNone(self.api.db.get_prompt_by_id(orphan))
+                self.assertEqual(self.backups(), [])
+                self.assertEqual(self.strays(), [])
+
+    async def test_policy_off_reports_no_backup(self):
+        from py.config import PromptManagerConfig
+
+        PromptManagerConfig.SAFETY_BACKUP = False
+        orphan = self.save("Orphan")
+        plan, _summary, job = await self.run_apply("unused_components")
+        self.assertEqual(plan["backup"]["policy"], "off")
+        self.assertEqual(job.state, "finished")
+        self.assertIn("No safety backup (the policy is off)", "\n".join(job.log))
+        self.assertEqual(
+            job.status_line, "Removed 1 unused prompt component · No safety backup"
+        )
+        self.assertEqual(self.backups(), [])
+        self.assertIsNone(self.api.db.get_prompt_by_id(orphan))
+
+    async def test_a_plan_reviewed_with_the_backup_on_keeps_that_promise(self):
+        from py.config import PromptManagerConfig
+
+        self.save("Orphan")
+        plan = await self.preview("unused_components")
+        PromptManagerConfig.SAFETY_BACKUP = False  # turned off after the review
+        body = await self.apply(plan["token"])
+        job = self.api._jobs.get(body["job"]["id"])
+        await asyncio.wait_for(asyncio.shield(job.task), WAIT)
+        self.assertEqual(len(self.backups()), 1)
+
+    async def test_backup_verified_before_write(self):
+        self.save("Orphan")
+        model, db = self.api.db.model, self.api.db
+        order = []
+        real_backup, real_verify = model.backup_database, model.verify_database_file
+        real_apply = db.apply_maintenance
+
+        def backup(path):
+            order.append("backup")
+            return real_backup(path)
+
+        def verify(path, **kwargs):
+            order.append("verify")
+            return real_verify(path, **kwargs)
+
+        def apply(*args):
+            order.append("apply")
+            self.assertEqual(len(self.backups()), 1)  # in place before the write
+            return real_apply(*args)
+
+        with (
+            patch.object(model, "backup_database", side_effect=backup),
+            patch.object(model, "verify_database_file", side_effect=verify),
+            patch.object(db, "apply_maintenance", side_effect=apply),
+        ):
+            _plan, _summary, job = await self.run_apply("unused_components")
+        self.assertEqual(job.state, "finished")
+        self.assertEqual(order, ["backup", "verify", "apply"])
+
+    async def test_missing_links_never_deletes_files(self):
+        prompt = self.save("Links")
+        kept = self.image(prompt, "kept.png")
+        gone = self.image(prompt, "sub/gone.png", exists=False)
+        other = Path(self.tmpdir) / "elsewhere"
+        self.image(prompt, "x.png", folder=other)
+        media = {
+            str(path): _digest(path)
+            for path in Path(self.tmpdir).rglob("*")
+            if path.is_file() and "prompts.db" not in path.name
+        }
+        _plan, _summary, job = await self.run_apply("missing_links")
+        self.assertEqual(job.state, "finished", job.log)
+        self.assertEqual(job.counters["changed"], 1)
+        self.assertEqual(
+            job.outcomes["changed"][0],
+            {
+                "path": "sub/gone.png",
+                "root": "output",
+                "root_index": 0,
+                "detail": "Link removed; the file was already missing",
+            },
+        )
+        after = {
+            str(path): _digest(path)
+            for path in Path(self.tmpdir).rglob("*")
+            if path.is_file() and "prompts.db" not in path.name and str(path) in media
+        }
+        self.assertEqual(after, media)
+        self.assertIsNone(self.api.db.get_image_by_id(gone))
+        self.assertIsNotNone(self.api.db.get_image_by_id(kept))
+
+    async def test_duplicate_prompts_and_optimize_apply(self):
+        kept = self.save("A cat")
+        with self.api.db.model.get_connection() as conn:
+            copy = conn.execute(
+                "INSERT INTO prompts(text, created_at) VALUES('A CAT', '2999-01-01')"
+            ).lastrowid
+        _plan, _summary, job = await self.run_apply("duplicate_prompts")
+        self.assertEqual(job.state, "finished", job.log)
+        self.assertEqual(job.status_line.split(" · ")[0], "Merged 1 duplicate prompt")
+        self.assertEqual(
+            job.outcomes["changed"][0]["detail"],
+            f"Prompt {copy} merged into prompt {kept}",
+        )
+        self.assertIsNone(self.api.db.get_prompt_by_id(copy))
+        _plan, _summary, job = await self.run_apply("optimize")
+        self.assertEqual(job.state, "finished", job.log)
+        self.assertEqual(job.status_line.split(" · ")[0], "Optimized the database")
+
+    async def test_protected_items_are_counted(self):
+        self.save("Orphan")
+        self.save("Kept", tags=["__protected__"])
+        _plan, _summary, job = await self.run_apply("unused_components")
+        self.assertEqual(job.counters, {"changed": 1, "protected": 1, "errors": 0})
+
+    async def test_apply_failure_changes_nothing_and_is_coded(self):
+        orphan = self.save("Orphan")
+        with patch.object(
+            self.api.db,
+            "apply_maintenance",
+            side_effect=sqlite3.OperationalError(f"locked {self.tmpdir}"),
+        ):
+            with self.assertLogs("prompt_manager", "ERROR"):
+                _plan, _summary, job = await self.run_apply("unused_components")
+        self.assertEqual(job.state, "failed")
+        self.assertEqual(job.last_event["code"], None)
+        self.assertEqual(job.status_line, "Maintenance failed; nothing was changed.")
+        self.assertNotIn(self.tmpdir, json.dumps(job.report()))
+        self.assertIsNotNone(self.api.db.get_prompt_by_id(orphan))
+
+    async def test_apply_errors_are_coded(self):
+        missing = await self.post(APPLY, {}, 400)
+        self.assertEqual(missing["code"], "invalid-request")
+        wrong = await self.post(APPLY, {"token": 7}, 400)
+        self.assertEqual(wrong["code"], "invalid-request")
+        extra = await self.post(APPLY, {"token": "x", "force": True}, 400)
+        self.assertEqual(extra["code"], "invalid-request")
+        expired = await self.apply("not-a-token", 410)
+        self.assertEqual(expired["code"], "plan-expired")
+
+    async def test_a_restore_after_the_review_is_storage_changed(self):
+        orphan = self.save("Orphan")
+        plan = await self.preview("unused_components")
+        self.api._database_replaced()  # what the legacy restore calls
+        reply = await self.apply(plan["token"], 409)
+        self.assertEqual(reply["code"], "storage-changed")
+        again = await self.apply(plan["token"], 410)  # the plan is forgotten
+        self.assertEqual(again["code"], "plan-expired")
+        self.assertIsNotNone(self.api.db.get_prompt_by_id(orphan))
+
+    async def test_restart_loses_plans_for_apply(self):
+        plan = await self.preview("optimize")
+        self.api._maintenance_plans = type(self.api._maintenance_plans)()
+        reply = await self.apply(plan["token"], 410)
+        self.assertEqual(reply["code"], "plan-expired")
+
+    async def test_safety_backups_are_pruned(self):
+        from py.api.admin import MAX_SAFETY_BACKUPS
+
+        for index in range(MAX_SAFETY_BACKUPS):
+            old = os.path.join(
+                self.tmpdir, f"prompts.db.backup_20000101_0000{index:02d}"
+            )
+            with open(old, "wb") as handle:
+                handle.write(b"old")
+            os.utime(old, (1000 + index, 1000 + index))
+        self.save("Orphan")
+        await self.run_apply("unused_components")
+        backups = self.backups()
+        self.assertEqual(len(backups), MAX_SAFETY_BACKUPS)
+        self.assertNotIn("prompts.db.backup_20000101_000000", backups)
+
+
+class TestSafetyBackup(ApplyTestCase):
+    async def test_backup_names_unique_within_one_second(self):
+        from py.api import maintenance
+
+        frozen = datetime.datetime(2026, 10, 23, 12, 0, 0, 123456)
+        model = self.api.db.model
+        with patch.object(maintenance, "_now", return_value=frozen):
+            names = [maintenance.take_safety_backup(model) for _ in range(3)]
+        self.assertEqual(
+            names,
+            [
+                "prompts.db.backup_20261023_120000_123456",
+                "prompts.db.backup_20261023_120000_123456_1",
+                "prompts.db.backup_20261023_120000_123456_2",
+            ],
+        )
+        for name in names:
+            ok, _reason = model.verify_database_file(os.path.join(self.tmpdir, name))
+            self.assertTrue(ok, name)
+        self.assertEqual(self.strays(), [])
+
+    async def test_backup_is_private_on_posix(self):
+        from py.api import maintenance
+
+        name = maintenance.take_safety_backup(self.api.db.model)
+        if os.name != "nt":
+            mode = os.stat(os.path.join(self.tmpdir, name)).st_mode & 0o777
+            self.assertEqual(mode, 0o600)
+
+    async def test_a_failed_replace_leaves_no_file(self):
+        from py.api import maintenance
+
+        with patch.object(maintenance.os, "replace", side_effect=OSError("busy")):
+            with self.assertRaises(maintenance.SafetyBackupFailed):
+                maintenance.take_safety_backup(self.api.db.model)
+        self.assertEqual(self.backups(), [])
+        self.assertEqual(self.strays(), [])
+
+    async def test_a_backup_that_cannot_be_finished_leaves_no_file(self):
+        from py.api import maintenance
+
+        real_connect = sqlite3.connect
+        calls = []
+
+        def connect(*args, **kwargs):
+            calls.append(args[0])
+            if len(calls) == 2:  # the backup is written; finishing it fails
+                raise sqlite3.OperationalError("x")
+            return real_connect(*args, **kwargs)
+
+        with patch.object(maintenance.sqlite3, "connect", side_effect=connect):
+            with self.assertRaises(maintenance.SafetyBackupFailed) as failed:
+                maintenance.take_safety_backup(self.api.db.model)
+        self.assertEqual(str(failed.exception), "The backup could not be finished")
+        self.assertEqual(calls[0], calls[1])  # the temp copy, both times
+        self.assertEqual((self.backups(), self.strays()), ([], []))
+
+    async def test_a_safety_backup_restores_through_the_legacy_route(self):
+        from aiohttp import FormData
+
+        from py.api import maintenance
+
+        self.save("Before the backup")
+        name = maintenance.take_safety_backup(self.api.db.model)
+        self.save("After the backup")
+        with open(os.path.join(self.tmpdir, name), "rb") as handle:
+            upload = handle.read()
+        form = FormData()
+        form.add_field("database_file", upload, filename=name)
+        response = await self.client.request(
+            "POST", "/prompt_manager/restore", data=form
+        )
+        self.assertEqual(response.status, 200, await response.text())
+        texts = [p["text"] for p in self.api.db.search_prompts()]
+        self.assertEqual(texts, ["Before the backup"])
+        mode = self.api.db.model.get_connection().execute("PRAGMA journal_mode")
+        self.assertEqual(mode.fetchone()[0], "wal")
+
+
+class TestSafetyBackupPolicy(MaintenanceAPITestCase):
+    async def test_policy_round_trips_with_revision(self):
+        settings = await self.client.request("GET", "/prompt_manager/settings")
+        body = await settings.json()
+        self.assertTrue(body["settings"]["safety_backup"])
+        saved = await self.post(
+            "/prompt_manager/settings",
+            {"safety_backup": False, "expected_revision": body["revision"]},
+            200,
+        )
+        self.assertTrue(saved["persisted"])
+        self.assertNotEqual(saved["revision"], body["revision"])
+        with open(self.config_path, encoding="utf-8") as handle:
+            self.assertIs(json.load(handle)["maintenance"]["safety_backup"], False)
+        again = await (
+            await self.client.request("GET", "/prompt_manager/settings")
+        ).json()
+        self.assertFalse(again["settings"]["safety_backup"])
+        self.assertEqual(again["revision"], saved["revision"])
+        plan = await self.preview("optimize")
+        self.assertEqual(plan["backup"]["policy"], "off")
+        stale = await self.post(
+            "/prompt_manager/settings",
+            {"safety_backup": True, "expected_revision": body["revision"]},
+            409,
+        )
+        self.assertEqual(stale["code"], "settings-changed")
+        bad = await self.post("/prompt_manager/settings", {"safety_backup": "yes"}, 400)
+        self.assertFalse(bad["success"])
+
+    async def test_config_file_value_is_read_at_load(self):
+        from py.config import PromptManagerConfig
+
+        PromptManagerConfig.update_config({"maintenance": {"safety_backup": False}})
+        self.assertFalse(PromptManagerConfig.SAFETY_BACKUP)
+        PromptManagerConfig.update_config({"maintenance": {"safety_backup": "no"}})
+        self.assertFalse(PromptManagerConfig.SAFETY_BACKUP)  # non-bool ignored
+        self.assertEqual(
+            PromptManagerConfig.get_config()["maintenance"], {"safety_backup": False}
+        )
+
+
 class TestLegacyMaintenance(MaintenanceAPITestCase):
     async def test_vacuum_failure_reported_not_success(self):
         with patch(
diff --git a/tests/test_maintenance_plans.py b/tests/test_maintenance_plans.py
index b5efe347..4d8533f7 100644
--- a/tests/test_maintenance_plans.py
+++ b/tests/test_maintenance_plans.py
@@ -428,6 +428,109 @@ class TestPreviews(MaintenanceDatabaseCase):
         # The teardown's assert_closed proves the read-only handle was closed.
 
 
+class TestApply(MaintenanceDatabaseCase):
+    """``apply_maintenance`` rechecks the fingerprint inside BEGIN IMMEDIATE."""
+
+    def apply(self, draft, roots=None):
+        roots = [str(self.output)] if roots is None else roots
+        return self.db.apply_maintenance(
+            draft.operation, draft.fingerprint, draft.targets, roots
+        )
+
+    def test_unused_components_apply_removes_exactly_the_reviewed_prompts(self):
+        first, second = self.save("One"), self.save("Two")
+        draft = self.preview("unused_components")
+        with patch.object(maintenance, "ID_BATCH", 1):  # one id per statement
+            result = self.apply(draft)
+        self.assertEqual(result["changed"], 2)
+        self.assertEqual(
+            [row["prompt_id"] for row in result["rows"]], sorted([first, second])
+        )
+        self.assertIsNone(self.db.get_prompt_by_id(first))
+        self.assertIsNone(self.db.get_prompt_by_id(second))
+
+    def test_changed_targets_raise_before_anything_is_written(self):
+        orphan, protected = self.save("Orphan"), self.save("Soon protected")
+        prompt = self.save("Links")
+        gone = self.image(prompt, "gone.png", exists=False)
+        kept = self.save("Kept")
+        self.duplicate("KEPT")
+        cases = {
+            "unused_components": lambda: self.db.model.get_connection().execute(
+                "INSERT INTO prompt_library_state(prompt_id,title) VALUES(?,?)",
+                (protected, "Now a recipe"),
+            ),
+            "missing_links": lambda: (self.output / "gone.png").write_bytes(b"back"),
+            "duplicate_prompts": lambda: self.db.update_prompt_metadata(kept, rating=4),
+        }
+        for operation, change in cases.items():
+            with self.subTest(operation=operation):
+                draft = self.preview(operation)
+                self.assertTrue(draft.targets)
+                change()
+                self.db.model.get_connection().commit()
+                with self.assertRaises(maintenance.PlanChanged):
+                    self.apply(draft)
+        self.assertIsNotNone(self.db.get_prompt_by_id(orphan))
+        self.assertIsNotNone(self.db.get_image_by_id(gone))
+        self.assertEqual(len(self.db.search_prompts(text="kept")), 2)
+
+    def test_a_link_whose_source_went_away_is_never_removed(self):
+        folder = self.root / "drive"
+        prompt = self.save("Links")
+        link = self.image(prompt, "gone.png", folder=folder, exists=False)
+        roots = [str(folder)]
+        draft = self.preview("missing_links", roots)
+        self.assertEqual(draft.targets, [link])
+        folder.rmdir()  # unmounted between the review and the apply
+        with self.assertRaises(maintenance.PlanChanged):
+            self.apply(draft, roots)
+        self.assertIsNotNone(self.db.get_image_by_id(link))
+
+    def test_missing_links_apply_removes_rows_only(self):
+        prompt = self.save("Links")
+        kept = self.image(prompt, "kept.png")
+        gone = [self.image(prompt, f"gone-{n}.png", exists=False) for n in range(3)]
+        media = {p.name: _digest(p) for p in self.output.iterdir()}
+        draft = self.preview("missing_links")
+        with patch.object(maintenance, "ID_BATCH", 2):
+            result = self.apply(draft)
+        self.assertEqual(result["changed"], 3)
+        self.assertEqual(
+            [os.path.basename(row["image_path"]) for row in result["rows"]],
+            [f"gone-{n}.png" for n in range(3)],
+        )
+        self.assertEqual({p.name: _digest(p) for p in self.output.iterdir()}, media)
+        self.assertIsNotNone(self.db.get_image_by_id(kept))
+        for image_id in gone:
+            self.assertIsNone(self.db.get_image_by_id(image_id))
+
+    def test_duplicate_prompts_apply_merges_into_the_oldest(self):
+        kept = self.save("A cat")
+        copy = self.duplicate("A CAT")
+        image = self.image(copy, "cat.png")
+        draft = self.preview("duplicate_prompts")
+        result = self.apply(draft)
+        self.assertEqual(result["changed"], 1)
+        self.assertEqual(result["rows"], [{"prompt_id": copy, "kept_id": kept}])
+        self.assertIsNone(self.db.get_prompt_by_id(copy))
+        self.assertEqual(self.db.get_image_by_id(image)["prompt_id"], kept)
+
+    def test_optimize_apply_vacuums(self):
+        self.save("Some")
+        draft = self.preview("optimize")
+        with patch.object(self.db.model, "vacuum_database") as vacuum:
+            self.assertEqual(self.apply(draft), {"changed": 1, "rows": []})
+        vacuum.assert_called_once_with()
+        self.assertEqual(self.apply(self.preview("optimize"))["changed"], 1)
+        with self.assertRaises(maintenance.PlanChanged):
+            self.db.apply_maintenance("optimize", "sha256:other", [], [])
+
+    def test_apply_of_an_unknown_operation_raises(self):
+        with self.assertRaises(ValueError):
+            self.db.apply_maintenance("reboot", "x", [], [])
+
+
 class TestVacuum(MaintenanceDatabaseCase):
     def test_vacuum_failure_raises(self):
         with patch(