feat(backup): restore from Settings → Backups, with restoring mode (#5816, #5817)

#5816 "Restoring mode": while a backup goes back in, the scheduler holds,
ingest and HTMX answer 503 with Retry-After (agents keep their samples),
and every page says what is happening. A safety backup of Steward as it
stands is made first, to /data/backups and the bucket. The backup's key
is put in place, the restore audited, and Steward exits with code 3 so
Swarm, compose or systemd starts it fresh. A failed in-place restore
changes nothing and says why on Backups; a full replace that fails after
the wipe restarts into first-run setup.

#5817 "Settings → Backups: Restore": a Restore button on each sent backup
and on the file card. The chosen backup is checked (the same comparison
as Check, now a shared partial) and held server-side for an hour under a
token; it starts only once the admin types "restore", and an older
backup also needs "Replace everything" ticked. A newer backup, or one
whose key can't open its secrets, gets no button.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

steward/settings/routes.py

@@ -695,7 +695,7 @@ async def send_report_now():
 async def _backups_page(flash=None):
     """Settings → Backups (#5673): how the newest backup went, the bucket and
     the daily hour (in the reader's zone), and the runs before."""
-    from steward.backup import job
+    from steward.backup import job, restoring
     from steward.core.formatting import zone
     async with current_app.db_sessionmaker() as db:
         settings = await get_all_settings(db)
@@ -705,7 +705,8 @@ async def _backups_page(flash=None):
         "settings/backups.html", settings=settings, flash=flash, shown_hour=hour,
         verdict=job.verdict(settings, job.is_running()),
         history=job.history_rows(settings.get("backup.history")),
-        has_secret=bool(settings.get("backup.s3_secret_key")))
+        has_secret=bool(settings.get("backup.s3_secret_key")),
+        restore_failed=restoring.STATE if restoring.STATE.error else None)
 
 
 @settings_bp.get("/backups/")
@@ -824,6 +825,108 @@ async def check_uploaded_backup():
     return await _check_page(report=report, name=name)
 
 
+
+# Restore (#5817): choose a backup, see what it would change (the same check as
+# above), then type to confirm. The chosen file is held server-side between
+# the two (restoring.stage), so the confirmation names it by a token.
+
+RESTORE_WORD = "restore"
+
+
+async def _restore_page(report=None, error: str = "", name: str = "", held=None):
+    return await render_template("settings/backup_restore.html", report=report, error=error,
+                                 name=name, held=held, word=RESTORE_WORD)
+
+
+async def _hold_and_check(held, name: str):
+    from steward.backup import job
+    from steward.backup.dump import BackupError
+    try:
+        report = await job.check_file(current_app._get_current_object(), held.path, name)
+    except BackupError as exc:
+        import shutil
+        shutil.rmtree(held.dir, ignore_errors=True)
+        return await _restore_page(error=str(exc), name=name)
+    held.name, held.schema, held.ok = name, report.schema, report.ok
+    held.save()
+    return await _restore_page(report=report, name=name, held=held)
+
+
+@settings_bp.post("/backups/restore/")
+@require_role(UserRole.admin)
+async def restore_stored_backup():
+    """Fetch one of the bucket's backups, check it, and ask to confirm."""
+    import shutil
+
+    from steward.backup import job, restoring
+    from steward.backup.dump import BackupError
+    from steward.backup.s3 import S3Error, get_file
+    name = (await request.form).get("name", "")
+    if not job.is_backup_name(name):
+        return await _restore_page(error=f"{name!r} isn't the name of a Steward backup.", name=name)
+    async with current_app.db_sessionmaker() as db:
+        target = job.target_from(await get_all_settings(db))
+    if target is None:
+        return await _restore_page(error="The bucket isn't set up: fill in the endpoint, bucket and keys.",
+                                   name=name)
+    held = restoring.stage()
+    try:
+        await get_file(target, name, held.path)
+    except (BackupError, S3Error) as exc:
+        shutil.rmtree(held.dir, ignore_errors=True)
+        error = str(exc)
+        if "AccessDenied" in error:
+            error += (" The bucket's key may not read backups back; download the file from the "
+                      "bucket and restore it with Restore a file instead.")
+        return await _restore_page(error=error, name=name)
+    return await _hold_and_check(held, name)
+
+
+@settings_bp.post("/backups/restore/file/")
+@require_role(UserRole.admin)
+async def restore_uploaded_backup():
+    """A backup file sent from the browser, checked and held for the
+    confirmation. Its size limit is raised in auth/csrf.py UPLOAD_LIMITS."""
+    from steward.backup import restoring
+    upload = (await request.files).get("file")
+    if upload is None or not upload.filename:
+        return await _restore_page(error="Choose a backup file to restore.")
+    held = restoring.stage()
+    await upload.save(held.path)
+    return await _hold_and_check(held, Path(upload.filename).name)
+
+
+@settings_bp.post("/backups/restore/start/")
+@require_role(UserRole.admin)
+async def start_restore():
+    """The confirmation: the held backup goes back in (restoring.py). Every
+    page shows the restore until Steward restarts."""
+    import shutil
+
+    from steward.backup import job, restoring
+    form = await request.form
+    held = restoring.staged(form.get("token", ""))
+    if held is None:
+        return await _restore_page(error="That backup is no longer held (it is removed an hour after "
+                                         "it was checked). Choose it again.")
+    if held.schema == "newer" or not held.ok:
+        shutil.rmtree(held.dir, ignore_errors=True)
+        return await _restore_page(error="This backup can't be restored here; check it again to see why.",
+                                   name=held.name)
+    replace_all = held.schema == "older"
+    if form.get("confirm", "").strip().lower() != RESTORE_WORD or (
+            replace_all and "replace_all" not in form):
+        report = await job.check_file(current_app._get_current_object(), held.path, held.name)
+        return await _restore_page(report=report, name=held.name, held=held,
+                                   error=f"Type {RESTORE_WORD} to confirm"
+                                         + (", and tick Replace everything." if replace_all else "."))
+    started = restoring.start_restore(current_app._get_current_object(), held.path, held.name,
+                                      username=session.get("username", "") or "an admin",
+                                      replace_all=replace_all, owned_dir=held.dir)
+    if not started:
+        return await _restore_page(error="A restore is already under way.", name=held.name)
+    return redirect(url_for("settings.backups"))
+
 def _build_plugin_cfg_from_form(plugin: dict, form) -> dict:
     """Extract config values for one plugin from a form submission."""
     name = plugin["_dir"]