Scheduled backups to a folder or S3 bucket, and forge tokens sealed at rest (milestone 523)

alembic/versions/0122_seal_forge_tokens.py

@@ -0,0 +1,82 @@
+"""Seal forge-connection tokens at rest (milestone 523)
+
+Revision ID: 0122
+Revises: 0121
+Create Date: 2026-10-15
+
+A forge token was stored as plain text, so a copy of the database (or of a
+backup tool's snapshot of it) was enough to read every user's forge
+credential. Tokens are now sealed the way the backup destination's S3 secret
+is (services/secrets.py): Fernet under a key derived from SECRET_KEY, which
+lives in the deployment's environment rather than in the data.
+
+The column is renamed `token` -> `token_sealed` in the same step. The name
+says what the bytes are, and every reader has to change with it, so a reader
+missed here fails on a missing attribute instead of quietly sending
+ciphertext to the forge as a token.
+
+THE KEY DERIVATION IS FROZEN HERE, not imported. This revision must seal with
+exactly v1's key forever; importing services.secrets would let a later edit
+there change what an old migration writes. The unit lane checks that what
+this produces is what `secrets.unseal` opens today.
+
+Cost: one row per (user, forge host) — a handful per instance.
+
+Downgrade opens each token back into plain text. A row sealed under a
+different SECRET_KEY cannot be opened and comes back empty; its owner enters
+the token again.
+"""
+import base64
+import hashlib
+
+import sqlalchemy as sa
+from alembic import op
+
+revision = "0122"
+down_revision = "0121"
+branch_labels = None
+depends_on = None
+
+_LABEL_V1 = b"scribe.secrets.v1\x00"
+
+
+def _fernet():
+    from cryptography.fernet import Fernet
+
+    from scribe.config import Config
+
+    digest = hashlib.sha256(_LABEL_V1 + Config.SECRET_KEY.encode()).digest()
+    return Fernet(base64.urlsafe_b64encode(digest))
+
+
+def seal_v1(plain: str) -> str:
+    return _fernet().encrypt(plain.encode()).decode()
+
+
+def upgrade() -> None:
+    op.alter_column("forge_connections", "token", new_column_name="token_sealed")
+    bind = op.get_bind()
+    rows = bind.execute(sa.text("SELECT id, token_sealed FROM forge_connections")).all()
+    for row_id, plain in rows:
+        bind.execute(
+            sa.text("UPDATE forge_connections SET token_sealed = :v WHERE id = :id"),
+            {"v": seal_v1(plain or ""), "id": row_id},
+        )
+
+
+def downgrade() -> None:
+    from cryptography.fernet import InvalidToken
+
+    bind = op.get_bind()
+    fernet = _fernet()
+    rows = bind.execute(sa.text("SELECT id, token_sealed FROM forge_connections")).all()
+    for row_id, sealed in rows:
+        try:
+            plain = fernet.decrypt((sealed or "").encode()).decode()
+        except (InvalidToken, ValueError):
+            plain = ""
+        bind.execute(
+            sa.text("UPDATE forge_connections SET token_sealed = :v WHERE id = :id"),
+            {"v": plain, "id": row_id},
+        )
+    op.alter_column("forge_connections", "token_sealed", new_column_name="token")

docs/backups.md

@@ -0,0 +1,160 @@
+# Backups
+
+Scribe can back itself up every day to a **folder** or an **S3-compatible
+bucket**. Each run adds one file; Scribe never deletes an old one. Keeping and
+thinning the history is the destination's job.
+
+Set it up under **Settings → Data → Scheduled backups** (admins only).
+
+## What a backup is
+
+One file per run, named for the moment it was taken in UTC:
+
+```
+scribe-20261015T020000Z.tar
+```
+
+It is a plain, **uncompressed** tar:
+
+- `manifest.json` comes first. It holds the backup format version, the Scribe
+  version, when the backup was taken, and how many rows each table has.
+- Then there is one `<table>.jsonl` per table, one row per line.
+
+It holds everything the admin **Full Backup** download holds, laid out
+differently. It does **not** hold:
+
+- **Search indexes.** They are rebuilt from the records when Scribe starts.
+- **Sharing:** groups, project shares and note shares.
+- **Credentials:** API keys and forge connections. After a restore, issue new
+  API keys and add forge connections again. (Forge tokens are also sealed under
+  `SECRET_KEY` in the database itself, so a copy of the database doesn't carry
+  them in readable form either.)
+- **Logs and notifications.**
+- **The backup destination's own S3 secret.**
+
+The file contains hashed passwords, so store it somewhere only trusted people
+can read.
+
+### Why it is shaped this way
+
+The file is meant to be copied off-site by a deduplicating tool such as
+restic, which stores a chunk of data only once no matter how many backups
+contain it. Every choice serves that:
+
+- **It is never compressed.** Compression changes every byte after the first
+  edit, so nothing would match the day before. restic compresses its own
+  repository after deduplicating, so nothing is lost.
+- **Rows are in primary-key order, with their keys sorted.** A row that did
+  not change is the same bytes in the same place every day. An edited row
+  changes one line, and new rows land at the end of their table's file.
+- **Each table is its own file**, so a busy table never shifts the bytes of a
+  quiet one.
+- **The parts that change every run are kept in `manifest.json`**: the time
+  and the counts.
+- **Each run gets a new, dated name.** Overwriting one name would keep one
+  copy and no history.
+
+Day to day, the off-site copy then stores roughly what changed, not the whole
+file again.
+
+## Destinations
+
+### A folder
+
+A path **inside the Scribe container**, normally a share mounted there:
+
+```yaml
+services:
+  scribe:
+    volumes:
+      - /nfs/backup/scribe:/backups
+```
+
+Scribe writes each file under a hidden name (`.scribe-….tar.part`) and renames
+it once it is complete. A program watching the folder never sees half a
+backup.
+
+### An S3-compatible bucket
+
+You need four things:
+
+- **Address:** the service itself, like `https://backup.example.com`. Don't
+  include the bucket; Scribe addresses buckets by path.
+- **Bucket.**
+- **Access key.**
+- **Secret key.**
+
+Scribe uploads and lists; it never deletes. A key that may **only add
+objects** is therefore enough, and it is the safer key to hand out: a stolen
+one cannot wipe the history.
+
+The secret is stored encrypted, under a key derived from `SECRET_KEY`. It is
+never shown again, and never included in a backup. If an instance's
+`SECRET_KEY` changes, for example after a restore onto a new instance, the
+page asks for the secret again.
+
+**Test destination** checks the destination without sending a backup:
+
+- a folder: it writes and removes a hidden file;
+- a bucket: it only lists, because an add-only key could never remove a test
+  file.
+
+## Schedule and history
+
+- Backups run once a day at the hour you choose, in UTC. The default is 02:00,
+  before the nightly trash purge and database maintenance.
+- **Back up now** runs one straight away.
+- The page leads with when the last good backup was taken. It says
+  **Overdue** when backups are on and none has worked for 26 hours.
+- It lists the last 30 runs, and gives the reason for each failure in words.
+- A run that hasn't finished after an hour is abandoned and recorded as
+  failed.
+
+## Example: Fabled Archives
+
+[Fabled Archives](https://git.fabledsword.com/bvandeusen/fabledarchives)
+keeps service backups as "backup targets". Either destination works with it.
+
+1. On Archives' **Backups** page, make a target for Scribe, named `scribe`
+   for example.
+2. Choose how Scribe's files arrive:
+   - **S3**: turn on the S3 way in for the target and make an **append-only**
+     S3 key. In Scribe, choose **An S3 bucket**:
+     - address: Archives' backup address (`http://<archives>:8081`);
+     - bucket: the target's name;
+     - access key and secret key: the key you just made.
+   - **A folder**: mount the share Archives watches into Scribe as
+     `/backups/scribe`, and point the target at that folder. In Scribe, choose
+     **A folder** with `/backups/scribe`.
+3. Set the target's **Retention** to **Archives**. Archives then keeps the
+   newest backups and thins the rest by day, week and month. Scribe only adds.
+4. Give the target an expected cadence of **daily**, so a missed backup turns
+   it late in Archives as well as overdue in Scribe.
+5. Add an off-site **restic** replication job for the target. The uncompressed
+   layout above is what lets that job send only what changed each day.
+
+## Restoring
+
+### Into a fresh instance
+
+Start Scribe on an empty database, copy the backup file into the container,
+and run:
+
+```sh
+docker exec -it scribe python -m scribe.restore /backups/scribe-20261015T020000Z.tar
+```
+
+- The command refuses a database that already has an account. A restore
+  re-creates everything with new ids, so on a live instance it would add a
+  second copy instead of replacing anything.
+- Nothing limits the file's size.
+- It also reads the JSON from the admin **Full Backup** download.
+
+Afterwards, **restart Scribe** so the search indexes are rebuilt. Then issue
+new API keys, add forge connections again, and set up any sharing.
+
+### From the web
+
+**Settings → Data → Restore from Backup** takes the `.tar` as well as the
+JSON download. The upload is held to the server's request-size limit. For a
+bigger file, use the command above.

docs/configuration.md

@@ -9,7 +9,7 @@ Configuration is via environment variables. The `docker-compose.yml` file sets d
 | Variable | Default | Description |
 |----------|---------|-------------|
 | `DATABASE_URL` | `postgresql+asyncpg://fabled:fabled@db/scribe` | PostgreSQL async connection string |
-| `SECRET_KEY` | `dev-secret-change-me` | Session signing key — **change this in production** |
+| `SECRET_KEY` | `dev-secret-change-me` | Session signing key, and the key stored credentials are sealed under — **change this in production, then keep it** |
 | `SECRET_KEY_FILE` | — | Path to a Docker secret file containing the key (alternative to `SECRET_KEY`) |
 | `LOG_LEVEL` | `INFO` | Logging verbosity (`DEBUG`, `INFO`, `WARNING`, `ERROR`) |
 | `SECURE_COOKIES` | `false` | Set `true` when running behind TLS |
@@ -127,6 +127,7 @@ If using `TRUST_PROXY_HEADERS=true`, ensure your proxy strips any client-supplie
   python3 -c "import secrets; print(secrets.token_hex(32))"
   ```
   Or use Docker Secrets via `SECRET_KEY_FILE`.
+- **Keep `SECRET_KEY` once it is set.** Scribe seals the credentials it stores under a key derived from it: forge-connection tokens and the backup destination's S3 secret. A copy of the database alone can't read them. If the key changes (for example, a restore onto an instance with a different key), those credentials can no longer be opened. Settings marks each one and asks for it again, and nothing else breaks.
 - **`SECURE_COOKIES=true`** — must be set when running behind TLS.
 - **`BASE_URL`** — set to your public URL; required for OIDC redirects and email links.
 - **Registration** — auto-closes after the first user (admin). Re-enable from Settings → Users or send invite links.

docs/features.md

@@ -127,8 +127,12 @@ Every read and mutation is scoped by owner + direct shares + group shares.
 
 - **Personal export** — download all your notes/tasks as a Markdown ZIP (YAML
   frontmatter) or a JSON array.
-- **Admin backup** — full application backup/restore (format v4) with ID remapping on
+- **Admin backup** — full application backup/restore with ID remapping on
   restore for cross-instance migration.
+- **Scheduled backups** — a daily, dated, uncompressed `.tar` to a folder or an
+  S3-compatible bucket, laid out so a deduplicating tool such as restic stores
+  only what changed; `python -m scribe.restore` brings a fresh instance back
+  from one. See [Backups](backups.md).
 
 ## Progressive Web App
 
@@ -148,7 +152,7 @@ reset, and session invalidation are included. See [SSO / OAuth](sso-oauth.md).
 | Account | Email change, password change, session invalidation |
 | Profile | Per-user profile fields |
 | Integrations | SearXNG status |
-| Data | Personal export; backup / restore (admin) |
+| Data | Personal export; backup / restore and scheduled backups (admin) |
 | API Keys | Create/revoke `fmcp_` keys for the MCP endpoint |
 | Config (admin) | Base URL, SMTP, DB-maintenance schedule |
 | Users (admin) | User list, invite links, registration toggle |

frontend/src/api/backups.ts

@@ -0,0 +1,73 @@
+import { apiGet, apiPost, apiPut } from "@/api/client";
+
+/**
+ * Scheduled backups (milestone 523) — `/api/admin/backups`. One dated,
+ * uncompressed tar per run, delivered to a folder or an S3 bucket.
+ */
+export interface BackupConfig {
+  enabled: boolean;
+  /** UTC hour the daily backup runs at, 0–23. */
+  hour: number;
+  destination: "folder" | "s3";
+  folder_path: string;
+  s3_endpoint: string;
+  s3_bucket: string;
+  s3_access_key: string;
+  s3_region: string;
+  s3_verify_tls: boolean;
+}
+
+export interface BackupRun {
+  started_at: string;
+  finished_at: string;
+  trigger: "scheduled" | "manual";
+  status: "ok" | "failed";
+  destination?: string;
+  file?: string;
+  size_bytes?: number;
+  where?: string;
+  error?: string;
+  duration_s: number;
+}
+
+export interface BackupsOverview {
+  config: BackupConfig;
+  /** Never the secret itself: whether one is kept, and whether this
+   *  instance can still open it. */
+  secret: { set: boolean; readable: boolean };
+  /** What stops the saved config from sending a backup, in words. */
+  problems: string[];
+  running: boolean;
+  /** Newest first, at most 30. */
+  runs: BackupRun[];
+  last_run: BackupRun | null;
+  last_success: BackupRun | null;
+  overdue: boolean;
+}
+
+/** What the form sends: config fields, plus the secret when one was typed. */
+export type BackupChanges = Partial<BackupConfig> & {
+  s3_secret?: string;
+  clear_s3_secret?: boolean;
+};
+
+export function getBackups(): Promise<BackupsOverview> {
+  return apiGet<BackupsOverview>("/api/admin/backups");
+}
+
+export function saveBackups(changes: BackupChanges): Promise<BackupsOverview> {
+  return apiPut<BackupsOverview>("/api/admin/backups", changes);
+}
+
+/** Ask the destination whether it would take a backup, with the form's
+ *  unsaved values. The server gives the destination 45s; this waits a little
+ *  longer so its own answer, not the client's deadline, is what arrives. */
+export function checkBackupDestination(
+  changes: BackupChanges,
+): Promise<{ ok: boolean; message: string }> {
+  return apiPost<{ ok: boolean; message: string }>("/api/admin/backups/check", changes, { timeoutMs: 60_000 });
+}
+
+export function runBackupNow(): Promise<{ status: string }> {
+  return apiPost<{ status: string }>("/api/admin/backups/run", {});
+}

frontend/src/components/BackupScheduleSettings.vue

@@ -0,0 +1,388 @@
+<script setup lang="ts">
+import { computed, onBeforeUnmount, onMounted, ref } from "vue";
+import { apiErrorMessage } from "@/api/client";
+import {
+  checkBackupDestination, getBackups, runBackupNow, saveBackups,
+  type BackupChanges, type BackupConfig, type BackupRun, type BackupsOverview,
+} from "@/api/backups";
+import { relativeTime } from "@/composables/useRelativeTime";
+import { formatBytes } from "@/utils/bytes";
+import { fmtStamp } from "@/utils/dateFormat";
+import { useToastStore } from "@/stores/toast";
+
+/**
+ * Scheduled backups (milestone 523): one dated, uncompressed tar per run,
+ * delivered to a folder or an S3 bucket. The page answers "is my data backed
+ * up?" first — the verdict, then anything wrong — and the settings follow.
+ */
+const toastStore = useToastStore();
+
+const data = ref<BackupsOverview | null>(null);
+const loading = ref(true);
+const loadError = ref("");
+const form = ref<BackupConfig | null>(null);
+// The secret is never sent back by the server; this holds only what the
+// admin types, and is sent only when non-empty.
+const secretInput = ref("");
+const saving = ref(false);
+const checking = ref(false);
+const checkResult = ref<{ ok: boolean; message: string } | null>(null);
+const starting = ref(false);
+const showAllRuns = ref(false);
+const RUNS_SHOWN = 8;
+
+let pollTimer: ReturnType<typeof setTimeout> | null = null;
+let unmounted = false;
+
+async function load() {
+  loadError.value = "";
+  try {
+    const fresh = await getBackups();
+    data.value = fresh;
+    // Keep the admin's unsaved edits; take the server's config only on the
+    // first load or right after a save.
+    if (!form.value) form.value = { ...fresh.config };
+  } catch (e: unknown) {
+    loadError.value = apiErrorMessage(e, "Could not load the backup settings");
+  } finally {
+    loading.value = false;
+    schedulePoll();
+  }
+}
+
+// While a backup runs, follow it. Re-armed from load()'s finally (rule 157),
+// so one failed read does not stop the page following the run.
+function schedulePoll() {
+  if (pollTimer) clearTimeout(pollTimer);
+  pollTimer = null;
+  if (!unmounted && data.value?.running) pollTimer = setTimeout(load, 3000);
+}
+
+onMounted(load);
+onBeforeUnmount(() => {
+  unmounted = true;
+  if (pollTimer) clearTimeout(pollTimer);
+});
+
+const hourLabel = (h: number) => `${String(h).padStart(2, "0")}:00 UTC`;
+
+// ── The verdict, and what is wrong ─────────────────────────────────────────
+const verdict = computed(() => {
+  const d = data.value;
+  if (!d) return { tone: "", text: "" };
+  if (d.running) return { tone: "", text: "Backing up now…" };
+  if (!d.config.enabled) {
+    return { tone: "muted", text: "Scheduled backups are off. Nothing is backed up unless you run one." };
+  }
+  const ok = d.last_success;
+  if (d.overdue) {
+    return {
+      tone: "error",
+      text: ok
+        ? `Overdue — the last backup that worked was ${relativeTime(ok.finished_at)} (${fmtStamp(ok.finished_at)}).`
+        : "Overdue — no backup has worked yet.",
+    };
+  }
+  if (!ok) return { tone: "", text: `On — the first backup runs at ${hourLabel(d.config.hour)}.` };
+  return {
+    tone: "ok",
+    text: `Backed up ${relativeTime(ok.finished_at)}: ${ok.file}, ${formatBytes(ok.size_bytes ?? 0)}, to ${ok.where}.`,
+  };
+});
+
+const lastFailure = computed(() => {
+  const last = data.value?.last_run;
+  return last && last.status === "failed" ? last : null;
+});
+
+const secretUnreadable = computed(() => {
+  const s = data.value?.secret;
+  return !!s && s.set && !s.readable;
+});
+
+const visibleRuns = computed<BackupRun[]>(() => {
+  const runs = data.value?.runs ?? [];
+  return showAllRuns.value ? runs : runs.slice(0, RUNS_SHOWN);
+});
+const hiddenRuns = computed(() => Math.max(0, (data.value?.runs.length ?? 0) - RUNS_SHOWN));
+
+// ── Actions ────────────────────────────────────────────────────────────────
+function changes(): BackupChanges {
+  const out: BackupChanges = { ...(form.value as BackupConfig) };
+  if (secretInput.value) out.s3_secret = secretInput.value;
+  return out;
+}
+
+async function save() {
+  if (!form.value) return;
+  saving.value = true;
+  try {
+    data.value = await saveBackups(changes());
+    form.value = { ...data.value.config };
+    secretInput.value = "";
+    toastStore.show("Backup settings saved");
+  } catch (e: unknown) {
+    toastStore.show(apiErrorMessage(e, "Could not save the backup settings"), "error");
+  } finally {
+    saving.value = false;
+  }
+}
+
+async function removeSecret() {
+  try {
+    data.value = await saveBackups({ clear_s3_secret: true });
+    toastStore.show("S3 secret removed");
+  } catch (e: unknown) {
+    toastStore.show(apiErrorMessage(e, "Could not remove the secret"), "error");
+  }
+}
+
+async function check() {
+  if (!form.value) return;
+  checking.value = true;
+  checkResult.value = null;
+  try {
+    checkResult.value = await checkBackupDestination(changes());
+  } catch (e: unknown) {
+    checkResult.value = { ok: false, message: apiErrorMessage(e, "The check could not be run") };
+  } finally {
+    checking.value = false;
+  }
+}
+
+async function backUpNow() {
+  starting.value = true;
+  try {
+    await runBackupNow();
+    if (data.value) data.value.running = true;
+    schedulePoll();
+  } catch (e: unknown) {
+    toastStore.show(apiErrorMessage(e, "Could not start a backup"), "error");
+  } finally {
+    starting.value = false;
+  }
+}
+</script>
+
+<template>
+  <div class="backups">
+    <p v-if="loading && !data" class="state">Loading…</p>
+    <p v-else-if="loadError && !data" class="state state-error" role="alert">
+      {{ loadError }}
+      <button type="button" class="btn-text" @click="load">Retry</button>
+    </p>
+
+    <template v-else-if="data && form">
+      <p :class="['verdict', `verdict-${verdict.tone}`]">{{ verdict.text }}</p>
+
+      <p v-if="lastFailure && !data.running" class="problem" role="alert">
+        The last run failed {{ relativeTime(lastFailure.finished_at) }}: {{ lastFailure.error }}
+      </p>
+      <p v-if="secretUnreadable" class="problem" role="alert">
+        The stored S3 secret can't be opened here — it was saved on an instance with a different
+        SECRET_KEY. Enter it again below.
+      </p>
+      <ul v-if="data.config.enabled && data.problems.length" class="problem-list" role="alert">
+        <li v-for="p in data.problems" :key="p">{{ p }}</li>
+      </ul>
+
+      <p class="intro">
+        Each run adds one dated file, <code>scribe-&lt;date&gt;.tar</code>, holding every table
+        as plain lines. It is not compressed, so a deduplicating tool such as restic stores only what
+        changed since the day before. Scribe never deletes old backups — the destination keeps and
+        thins the history. Sharing, API keys and forge connections are not included.
+      </p>
+
+      <div class="row">
+        <label class="check">
+          <input v-model="form.enabled" type="checkbox" />
+          Back up daily
+        </label>
+        <label class="inline-field">
+          at
+          <select v-model.number="form.hour" class="fs-input" aria-label="Backup hour (UTC)">
+            <option v-for="h in 24" :key="h - 1" :value="h - 1">{{ hourLabel(h - 1) }}</option>
+          </select>
+        </label>
+      </div>
+
+      <fieldset class="destination">
+        <legend>Send each backup to</legend>
+        <label class="check">
+          <input v-model="form.destination" type="radio" value="folder" />
+          A folder
+        </label>
+        <label class="check">
+          <input v-model="form.destination" type="radio" value="s3" />
+          An S3 bucket
+        </label>
+      </fieldset>
+
+      <div v-if="form.destination === 'folder'" class="fields">
+        <label class="field">
+          <span>Folder</span>
+          <input v-model.trim="form.folder_path" class="fs-input" placeholder="/backups" />
+          <span class="field-hint">
+            A path inside the Scribe container — usually a share mounted there that a backup
+            service watches.
+          </span>
+        </label>
+      </div>
+
+      <div v-else class="fields">
+        <label class="field">
+          <span>Address</span>
+          <input v-model.trim="form.s3_endpoint" class="fs-input" placeholder="https://backup.example.com" />
+          <span class="field-hint">The S3 service itself, without the bucket. Buckets are addressed by path.</span>
+        </label>
+        <label class="field">
+          <span>Bucket</span>
+          <input v-model.trim="form.s3_bucket" class="fs-input" placeholder="scribe" />
+        </label>
+        <label class="field">
+          <span>Access key</span>
+          <input v-model.trim="form.s3_access_key" class="fs-input" autocomplete="off" />
+        </label>
+        <label class="field">
+          <span>Secret key</span>
+          <input
+            v-model="secretInput"
+            type="password"
+            class="fs-input"
+            autocomplete="new-password"
+            :placeholder="data.secret.set && data.secret.readable ? 'Saved — type a new one to replace it' : ''"
+          />
+          <span class="field-hint">
+            Stored encrypted, and never included in a backup. A key that may only add files is enough.
+            <button
+              v-if="data.secret.set"
+              type="button"
+              class="btn-text"
+              @click="removeSecret"
+            >Remove the saved secret</button>
+          </span>
+        </label>
+        <label class="field">
+          <span>Region</span>
+          <input v-model.trim="form.s3_region" class="fs-input" placeholder="us-east-1" />
+        </label>
+        <label class="check">
+          <input v-model="form.s3_verify_tls" type="checkbox" />
+          Check the address's TLS certificate
+        </label>
+      </div>
+
+      <div class="actions">
+        <button type="button" class="btn-primary" :disabled="saving" @click="save">
+          {{ saving ? "Saving…" : "Save" }}
+        </button>
+        <button type="button" class="btn-secondary" :disabled="checking" @click="check">
+          {{ checking ? "Checking…" : "Test destination" }}
+        </button>
+        <button
+          type="button"
+          class="btn-secondary"
+          :disabled="starting || data.running"
+          @click="backUpNow"
+        >
+          {{ data.running ? "Backing up…" : "Back up now" }}
+        </button>
+      </div>
+      <p v-if="checkResult" :class="['check-result', checkResult.ok ? 'check-ok' : 'check-bad']" role="status">
+        {{ checkResult.message }}
+      </p>
+
+      <h3 class="runs-label">Recent runs</h3>
+      <p v-if="!data.runs.length" class="state">None yet.</p>
+      <div v-else class="runs-scroll">
+        <table class="runs">
+          <thead>
+            <tr><th>When</th><th>Result</th><th>File</th><th class="num">Size</th><th class="num">Took</th></tr>
+          </thead>
+          <tbody>
+            <tr v-for="r in visibleRuns" :key="r.started_at" :class="{ failed: r.status === 'failed' }">
+              <td :title="fmtStamp(r.started_at)">
+                {{ relativeTime(r.started_at) }}
+                <span v-if="r.trigger === 'manual'" class="by-hand">by hand</span>
+              </td>
+              <td>{{ r.status === "ok" ? "Backed up" : r.error }}</td>
+              <td><code v-if="r.file">{{ r.file }}</code><span v-else>—</span></td>
+              <td class="num">{{ r.size_bytes != null ? formatBytes(r.size_bytes) : "—" }}</td>
+              <td class="num">{{ r.duration_s }} s</td>
+            </tr>
+          </tbody>
+        </table>
+        <button v-if="hiddenRuns && !showAllRuns" type="button" class="btn-text" @click="showAllRuns = true">
+          +{{ hiddenRuns }} earlier
+        </button>
+      </div>
+    </template>
+  </div>
+</template>
+
+<style scoped>
+.backups { display: flex; flex-direction: column; gap: var(--fs-space-3); }
+.state { margin: 0; font-size: var(--fs-size-body-sm); color: var(--fs-text-secondary); }
+.state-error { color: var(--fs-error); }
+
+.verdict { margin: 0; font-size: var(--fs-size-body); color: var(--fs-text-primary); }
+.verdict-ok { color: var(--fs-success); }
+.verdict-error { color: var(--fs-error); }
+.verdict-muted { color: var(--fs-text-secondary); }
+.problem, .problem-list {
+  margin: 0;
+  font-size: var(--fs-size-body-sm);
+  color: var(--fs-error);
+}
+.problem-list { padding-left: var(--fs-space-4); }
+
+.intro { margin: 0; font-size: var(--fs-size-body-sm); color: var(--fs-text-secondary); line-height: var(--fs-leading-body); }
+.intro code, .runs code { font-family: var(--fs-font-mono); font-size: var(--fs-size-tiny); }
+
+.row { display: flex; flex-wrap: wrap; align-items: center; gap: var(--fs-space-3); }
+.check { display: inline-flex; align-items: center; gap: var(--fs-space-2); font-size: var(--fs-size-body-sm); color: var(--fs-text-primary); }
+.inline-field { display: inline-flex; align-items: center; gap: var(--fs-space-2); font-size: var(--fs-size-body-sm); color: var(--fs-text-secondary); }
+.inline-field select { width: auto; }
+
+.destination {
+  margin: 0; padding: 0; border: 0;
+  display: flex; flex-wrap: wrap; align-items: center; gap: var(--fs-space-3);
+}
+.destination legend {
+  float: left; margin-right: var(--fs-space-2);
+  font-size: var(--fs-size-body-sm); color: var(--fs-text-secondary);
+}
+
+.fields { display: flex; flex-direction: column; gap: var(--fs-space-3); max-width: 32rem; }
+.field { display: flex; flex-direction: column; gap: var(--fs-space-1); }
+.field > span:first-child { font-size: var(--fs-size-body-sm); color: var(--fs-text-secondary); }
+
+.actions { display: flex; flex-wrap: wrap; gap: var(--fs-space-2); }
+.check-result { margin: 0; font-size: var(--fs-size-body-sm); }
+.check-ok { color: var(--fs-success); }
+.check-bad { color: var(--fs-error); }
+
+.runs-label {
+  margin: var(--fs-space-2) 0 0;
+  font-size: var(--fs-size-tiny); font-weight: normal;
+  text-transform: uppercase; letter-spacing: var(--fs-tracking-tiny);
+  color: var(--fs-text-tertiary);
+}
+.runs-scroll { overflow-x: auto; }
+.runs { width: 100%; border-collapse: collapse; font-size: var(--fs-size-body-sm); }
+.runs th {
+  text-align: left; font-weight: normal;
+  font-size: var(--fs-size-tiny); color: var(--fs-text-tertiary);
+  padding: var(--fs-space-1) var(--fs-space-2);
+  border-bottom: 1px solid var(--fs-border-color);
+}
+.runs td {
+  padding: var(--fs-space-1) var(--fs-space-2);
+  color: var(--fs-text-primary);
+  border-bottom: 1px solid var(--fs-border-color);
+  vertical-align: baseline;
+}
+.runs .num { text-align: right; font-variant-numeric: tabular-nums; white-space: nowrap; }
+.runs tr.failed td { color: var(--fs-error); }
+.by-hand { margin-left: var(--fs-space-1); font-size: var(--fs-size-tiny); color: var(--fs-text-tertiary); }
+</style>

frontend/src/utils/bytes.ts

@@ -0,0 +1,8 @@
+/** "1.4 MB" — a byte count for people. Binary units, one decimal under 10. */
+export function formatBytes(n: number): string {
+  if (n < 1024) return `${n} B`;
+  const units = ["KB", "MB", "GB", "TB"];
+  let v = n / 1024, i = 0;
+  while (v >= 1024 && i < units.length - 1) { v /= 1024; i++; }
+  return `${v.toFixed(v >= 10 || i === 0 ? 0 : 1)} ${units[i]}`;
+}

frontend/src/views/SettingsView.vue

@@ -11,6 +11,8 @@ import PaginationBar from "@/components/PaginationBar.vue";
 import TagInput from "@/components/TagInput.vue";
 import MomentsSettings from "@/components/MomentsSettings.vue";
 import ReplyShapesSettings from "@/components/ReplyShapesSettings.vue";
+import BackupScheduleSettings from "@/components/BackupScheduleSettings.vue";
+import { formatBytes } from "@/utils/bytes";
 import { fmtDate, fmtLogStamp } from "@/utils/dateFormat";
 import { fetchVersion, type VersionPayload } from "@/api/version";
 
@@ -640,13 +642,6 @@ const dbHealth = ref<DbHealth | null>(null);
 const loadingHealth = ref(false);
 const DEAD_PCT_WARN = 20; // dead-tuple ratio above this = autovacuum falling behind
 
-function formatBytes(n: number): string {
-  if (n < 1024) return `${n} B`;
-  const units = ["KB", "MB", "GB", "TB"];
-  let v = n / 1024, i = 0;
-  while (v >= 1024 && i < units.length - 1) { v /= 1024; i++; }
-  return `${v.toFixed(v >= 10 || i === 0 ? 0 : 1)} ${units[i]}`;
-}
 const pluginInstallCommands = computed(() => {
   const mkt = (pluginMarketplaceUrl.value || '').trim()
     || serverMarketplaceUrl.value
@@ -830,14 +825,17 @@ const baseUrlSaved = ref(false);
 // Git forge connections (#2778) — the user's keyring: one read-only
 // credential per forge host, used server-side for every forge read on
 // projects this user owns. The token round-trips masked; the server treats
-// the mask as "unchanged".
+// the mask as "unchanged". Tokens are stored sealed under SECRET_KEY:
+// `token_readable` false means this instance can no longer open one (the key
+// changed since it was saved), so the row asks for it again.
 interface ForgeConnectionEntry {
-  id: number; kind: string; base_url: string; host: string;
+  id: number; kind: string; base_url: string; host: string; token_readable: boolean;
 }
 const forgeConnections = ref<ForgeConnectionEntry[]>([]);
 const forgeKinds = ref<string[]>(["gitea", "github"]);
 const connForm = ref({ id: 0, kind: "gitea", base_url: "", token: "" });
 const connFormOpen = ref(false);
+const unreadableConnections = computed(() => forgeConnections.value.filter((c) => !c.token_readable));
 const savingConn = ref(false);
 const testingConnId = ref(0);
 const connTestResult = ref<{ id: number; ok: boolean; message: string } | null>(null);
@@ -1368,14 +1366,21 @@ async function handleRestoreFile(event: Event) {
   if (!file) return;
   restoring.value = true;
   try {
-    const text = await file.text();
-    const data = JSON.parse(text);
+    // The scheduled backup's .tar goes up as it is; the JSON download is
+    // parsed first so a damaged file fails here, before the upload.
+    const isTar = file.name.toLowerCase().endsWith(".tar");
     const res = await fetch("/api/admin/restore", {
       method: "POST",
-      headers: { "Content-Type": "application/json" },
-      body: JSON.stringify(data),
+      headers: { "Content-Type": isTar ? "application/x-tar" : "application/json" },
+      body: isTar ? file : JSON.stringify(JSON.parse(await file.text())),
       signal: bulkDeadline(),
     });
+    if (res.status === 413) {
+      throw new Error(
+        "the file is bigger than the web upload allows. Restore it with " +
+        "`python -m scribe.restore <file>` inside the Scribe container instead.",
+      );
+    }
     if (!res.ok) {
       const body = await res.json().catch(() => ({ error: `Error ${res.status}` }));
       throw new Error((body as Record<string, string>).error || `Error ${res.status}`);
@@ -2562,13 +2567,23 @@ async function deleteUser(userId: number) {
           Read-only, base URL <code>https://github.com</code> (or your GitHub
           Enterprise URL).
         </p>
+        <p v-if="unreadableConnections.length" class="calibration-warning" role="alert">
+          {{ unreadableConnections.length === 1 ? "One connection's token" : `${unreadableConnections.length} connections' tokens` }}
+          can't be read on this instance: <code>SECRET_KEY</code> has changed since
+          {{ unreadableConnections.length === 1 ? "it was" : "they were" }} saved. Edit
+          {{ unreadableConnections.map((c) => c.host).join(", ") }} and enter the token again.
+          Until then, forge reads for that host don't run.
+        </p>
         <table v-if="forgeConnections.length" class="api-keys-table">
           <thead>
             <tr><th>Host</th><th>Kind</th><th>Base URL</th><th></th></tr>
           </thead>
           <tbody>
             <tr v-for="c in forgeConnections" :key="c.id">
-              <td>{{ c.host }}</td>
+              <td>
+                {{ c.host }}
+                <span v-if="!c.token_readable" class="conn-token-lost">token needs re-entering</span>
+              </td>
               <td>{{ c.kind }}</td>
               <td><code>{{ c.base_url }}</code></td>
               <td>
@@ -2650,12 +2665,16 @@ async function deleteUser(userId: number) {
             <input
               ref="restoreFileInput"
               type="file"
-              accept=".json"
+              accept=".json,.tar"
               class="hidden-file-input"
               @change="handleRestoreFile"
             />
           </div>
         </section>
+        <section class="settings-section full-width">
+          <h2>Scheduled backups</h2>
+          <BackupScheduleSettings />
+        </section>
       </template>
 
     </div>
@@ -4587,6 +4606,11 @@ async function deleteUser(userId: number) {
   border-radius: var(--fs-radius-md);
   background: color-mix(in srgb, var(--fs-warning) 12%, var(--fs-surface-raised));
 }
+.conn-token-lost {
+  margin-left: var(--fs-space-2);
+  font-size: var(--fs-size-body-sm);
+  color: var(--fs-warning-fg);
+}
 .calibration-warning-title {
   margin: 0 0 var(--fs-space-2);
   font-size: 0.95rem;

pyproject.toml

@@ -13,6 +13,10 @@ dependencies = [
     "asyncpg>=0.29",
     "alembic>=1.13",
     "httpx>=0.27",
+    # Seals the backup destination's S3 secret at rest (services/secrets.py).
+    # Already in the image through pyjwt; named here because Scribe now
+    # imports it directly.
+    "cryptography",
     "hypercorn>=0.17",
     "bcrypt>=4.0",
     "aiosmtplib>=3.0",

src/scribe/app.py

@@ -304,6 +304,11 @@ def create_app() -> Quart:
                 asyncio.get_running_loop(), await get_maintenance_hour()
             )
 
+            # Scheduled backups (milestone 523). An hourly tick that reads its
+            # config when it fires, so starting it reads nothing here (#4181).
+            from scribe.services.backup_job import start_backup_scheduler
+            start_backup_scheduler(asyncio.get_running_loop())
+
             # Diagnostic instrumentation — heartbeat, signal handlers, asyncio
             # exception hook. Cheap (~1 log line/min), high diagnostic value when
             # the app crashes mysteriously. See services/diagnostics.py.
@@ -330,6 +335,8 @@ def create_app() -> Quart:
             stop_db_maintenance_scheduler,
         )
         stop_db_maintenance_scheduler()
+        from scribe.services.backup_job import stop_backup_scheduler
+        stop_backup_scheduler()
         from scribe.services.diagnostics import stop_diagnostics
         stop_diagnostics()
 

src/scribe/models/forge_connection.py

@@ -16,7 +16,9 @@ class ForgeConnection(Base, TimestampMixin):
 
     `host` is derived from `base_url` at write time and stored because it is
     the lookup key; the service layer keeps the two in step. The token is a
-    secret: to_dict never includes it, and no route may return it.
+    secret: it is stored sealed (services/secrets.py — read it through
+    services.forge_connections.connection_token), to_dict never includes it,
+    and no route may return it.
     """
 
     __tablename__ = "forge_connections"
@@ -31,7 +33,7 @@ class ForgeConnection(Base, TimestampMixin):
     kind: Mapped[str] = mapped_column(Text, nullable=False)
     base_url: Mapped[str] = mapped_column(Text, nullable=False)
     host: Mapped[str] = mapped_column(Text, nullable=False)
-    token: Mapped[str] = mapped_column(Text, nullable=False)
+    token_sealed: Mapped[str] = mapped_column(Text, nullable=False)
 
     def to_dict(self) -> dict:
         return {

src/scribe/restore.py

@@ -0,0 +1,95 @@
+"""Restore a Scribe backup into an empty instance, from the command line.
+
+    python -m scribe.restore /backups/scribe-20261015T020000Z.tar
+
+This is how a fresh instance is brought back after losing the old one. Run it
+inside the Scribe container, which already has the database settings and has
+already migrated the schema on start. It reads either the scheduled backup's
+tar (services/backup_file.py) or the JSON the admin download produces.
+
+It refuses a database that already holds an account. A restore re-creates
+every user, project and record from the file with new ids; into a live
+instance it would add a second copy of everything rather than replace it.
+(Settings → Backup & Restore still offers the upload for that case, on
+purpose and with its own warning.)
+
+Unlike the upload, nothing limits the file's size here, so this is also the
+way in for a backup too big for the web upload.
+
+Afterwards, restart Scribe: search indexes are not part of a backup and are
+rebuilt from the restored records at startup.
+"""
+from __future__ import annotations
+
+import argparse
+import asyncio
+import json
+import sys
+from pathlib import Path
+
+from sqlalchemy import func, select
+
+
+def load_backup(path: Path) -> dict:
+    """The payload in `path`, whichever of the two backup formats it is."""
+    from scribe.services.backup_file import is_archive, read_archive
+
+    with open(path, "rb") as f:
+        if is_archive(f):
+            return read_archive(f)
+        return json.load(f)
+
+
+async def _account_count() -> int:
+    from scribe.models import async_session
+    from scribe.models.user import User
+
+    async with async_session() as session:
+        return (await session.execute(select(func.count()).select_from(User))).scalar_one()
+
+
+async def run(path: Path) -> int:
+    from scribe.models import engine
+    from scribe.services.backup import restore_full_backup
+    from scribe.services.backup_file import ArchiveError
+
+    try:
+        try:
+            payload = load_backup(path)
+        except FileNotFoundError:
+            print(f"There is no file at {path}.", file=sys.stderr)
+            return 2
+        except (ArchiveError, ValueError) as exc:
+            print(f"{path} isn't a backup this Scribe can read: {exc}", file=sys.stderr)
+            return 2
+
+        accounts = await _account_count()
+        if accounts:
+            print(
+                f"This database already holds {accounts} account(s). A restore adds a second "
+                "copy of everything rather than replacing it, so it only runs into an empty "
+                "instance. Start Scribe on an empty database and run this again.",
+                file=sys.stderr,
+            )
+            return 3
+
+        stats = await restore_full_backup(payload)
+        print(json.dumps({"restored": stats}, indent=2, default=str))
+        print("Restored. Restart Scribe so the search indexes are rebuilt.", file=sys.stderr)
+        return 0
+    finally:
+        await engine.dispose()
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(
+        prog="python -m scribe.restore",
+        description="Restore a Scribe backup (.tar or .json) into an empty instance.",
+    )
+    parser.add_argument("file", type=Path, help="the backup file")
+    args = parser.parse_args(argv)
+    return asyncio.run(run(args.file))
+
+
+if __name__ == "__main__":
+    sys.exit(main())

src/scribe/routes/admin.py

@@ -59,7 +59,21 @@ async def backup():
 @admin_bp.route("/restore", methods=["POST"])
 @admin_required
 async def restore():
-    data = await request.get_json()
+    """Restore from an uploaded backup: the JSON download (application/json)
+    or the scheduled backup's tar (sent as the raw body). The upload is held
+    to the app's request-size limit; `python -m scribe.restore` has none, and
+    is the way in for a bigger file or a fresh instance."""
+    if request.mimetype == "application/json":
+        data = await request.get_json()
+    else:
+        import io
+
+        from scribe.services.backup_file import ArchiveError, read_archive
+
+        try:
+            data = read_archive(io.BytesIO(await request.get_data()))
+        except ArchiveError as exc:
+            return jsonify({"error": f"That file isn't a Scribe backup: {exc}"}), 400
     if not data:
         return jsonify({"error": "No backup data provided"}), 400
 
@@ -315,6 +329,120 @@ async def run_db_maintenance_now():
     return jsonify(summary)
 
 
+# ── Scheduled backups (milestone 523) ────────────────────────────────────────
+# One dated, uncompressed tar per run to a folder or an S3 bucket — see
+# services/backup_file.py for why it is shaped that way, backup_destination.py
+# for where it goes, backup_job.py for when.
+
+# How long "Test" waits on the destination. A folder on a hung NFS mount has
+# no timeout of its own (rule 156), and the request must come back either way.
+_BACKUP_CHECK_TIMEOUT = 45.0
+
+
+async def _backups_payload() -> dict:
+    from datetime import datetime, timezone
+
+    from scribe.services import backup_destination as dest
+    from scribe.services import backup_job as job
+
+    config = await dest.load_config()
+    secret, stored = await dest.load_secret()
+    runs = await job.list_runs()
+    return {
+        "config": config.to_dict(),
+        # Never the secret itself: whether one is kept, and whether this
+        # instance can still open it (it can't after a restore onto another
+        # SECRET_KEY, and then it has to be entered again).
+        "secret": {"set": stored, "readable": secret is not None},
+        "problems": config.problems(has_secret=secret is not None),
+        "running": job.is_running(),
+        "runs": runs,
+        **job.summarise(runs, enabled=config.enabled, now=datetime.now(timezone.utc)),
+    }
+
+
+@admin_bp.route("/backups", methods=["GET"])
+@admin_required
+async def get_backups():
+    """The backup destination, its schedule, and the recent runs."""
+    return jsonify(await _backups_payload())
+
+
+@admin_bp.route("/backups", methods=["PUT"])
+@admin_required
+async def update_backups():
+    """Change the destination or schedule. `s3_secret` replaces the stored
+    secret when it is sent and isn't the mask; `clear_s3_secret` removes it."""
+    from scribe.services import backup_destination as dest
+
+    data = await request.get_json() or {}
+    try:
+        config = dest.apply_changes(await dest.load_config(), data)
+    except ValueError as exc:
+        return jsonify({"error": str(exc)}), 400
+    await dest.save_config(config)
+    secret = data.get("s3_secret")
+    if data.get("clear_s3_secret"):
+        await dest.save_secret("")
+    elif isinstance(secret, str) and secret and secret != SECRET_MASK:
+        await dest.save_secret(secret)
+    uid = get_current_user_id()
+    await log_audit(
+        "backup_config", user_id=uid, username=g.user.username,
+        ip_address=request.remote_addr,
+        details={**config.to_dict(), "secret_changed": bool(
+            data.get("clear_s3_secret") or (isinstance(secret, str) and secret and secret != SECRET_MASK)
+        )},
+    )
+    return jsonify(await _backups_payload())
+
+
+@admin_bp.route("/backups/check", methods=["POST"])
+@admin_required
+async def check_backups_destination():
+    """Ask the destination whether it would take a backup, without sending
+    one. The body may carry unsaved form values, so a destination can be
+    tested before it is saved; the stored secret is used unless one is sent."""
+    from scribe.services import backup_destination as dest
+
+    data = await request.get_json() or {}
+    try:
+        config = dest.apply_changes(await dest.load_config(), data)
+    except ValueError as exc:
+        return jsonify({"ok": False, "message": str(exc)})
+    secret = data.get("s3_secret")
+    if not (isinstance(secret, str) and secret and secret != SECRET_MASK):
+        secret, _ = await dest.load_secret()
+    try:
+        message = await asyncio.wait_for(
+            dest.check_destination(config, secret), _BACKUP_CHECK_TIMEOUT,
+        )
+    except dest.DestinationError as exc:
+        return jsonify({"ok": False, "message": str(exc)})
+    except (TimeoutError, asyncio.TimeoutError):
+        return jsonify({"ok": False, "message": (
+            f"The destination didn't answer within {int(_BACKUP_CHECK_TIMEOUT)} seconds."
+        )})
+    return jsonify({"ok": True, "message": message})
+
+
+@admin_bp.route("/backups/run", methods=["POST"])
+@admin_required
+async def run_backup_now():
+    """Start a backup now. It runs in the background; the page follows it
+    through GET /backups (`running`, then the newest run)."""
+    from scribe.services import backup_job as job
+
+    try:
+        job.start_backup_now()
+    except job.BackupBusy:
+        return jsonify({"error": "A backup is already running."}), 409
+    uid = get_current_user_id()
+    await log_audit("backup_run", user_id=uid, username=g.user.username,
+                    ip_address=request.remote_addr, details={})
+    return jsonify({"status": "started"}), 202
+
+
 @admin_bp.route("/invitations", methods=["POST"])
 @admin_required
 async def create_invite():

src/scribe/routes/settings.py

@@ -125,12 +125,17 @@ async def test_search():
 @login_required
 async def list_forge_connections_route():
     from scribe.services.forge import FORGE_KINDS
-    from scribe.services.forge_connections import list_connections
+    from scribe.services.forge_connections import connection_token, list_connections
 
     uid = get_current_user_id()
     rows = await list_connections(uid)
+    # Never the token: only whether this instance can still open it, so the
+    # page can ask for it again after a SECRET_KEY change.
     return jsonify({
-        "connections": [r.to_dict() for r in rows],
+        "connections": [
+            {**r.to_dict(), "token_readable": connection_token(r) is not None}
+            for r in rows
+        ],
         "kinds": list(FORGE_KINDS),
     })
 
@@ -198,13 +203,19 @@ async def test_forge_connection_route(connection_id: int):
     press, so a misconfiguration is visible now rather than as silent
     fallbacks later (#2663's lesson, applied per keyring row)."""
     from scribe.services.forge import ForgeError, build_adapter
-    from scribe.services.forge_connections import get_connection
+    from scribe.services.forge_connections import connection_token, get_connection
 
     uid = get_current_user_id()
     row = await get_connection(uid, connection_id)
     if row is None:
         return jsonify({"error": "Connection not found"}), 404
-    adapter = build_adapter(row.kind, row.base_url, row.token)
+    token = connection_token(row)
+    if token is None:
+        return jsonify({
+            "error": "This instance can't read the saved token: SECRET_KEY has changed "
+                     "since it was saved. Enter the token again."
+        }), 400
+    adapter = build_adapter(row.kind, row.base_url, token)
     if adapter is None:
         return jsonify({"error": "Connection is not usable — check kind and base URL"}), 400
     try:

src/scribe/services/backup.py

@@ -772,8 +772,22 @@ def _rule_version_rows(rows) -> list[dict]:
     ]
 
 
+# Setting KEYS no backup carries. The backup destination's S3 secret is sealed
+# under SECRET_KEY, but a backup is exactly the thing copied off-site — it must
+# not carry the key that writes to the place it is copied to. A restore leaves
+# the destination configured and asks for the secret again.
+#
+# `backup_runs` is the scheduled backup's own run history: a record of what
+# THIS instance sent where. Restored onto another instance it would claim
+# backups that instance never made.
+_SETTINGS_NOT_EXPORTED = frozenset({"backup_s3_secret", "backup_runs"})
+
+
 def _setting_rows(rows) -> list[dict]:
-    return [{"user_id": s.user_id, "key": s.key, "value": s.value} for s in rows]
+    return [
+        {"user_id": s.user_id, "key": s.key, "value": s.value}
+        for s in rows if s.key not in _SETTINGS_NOT_EXPORTED
+    ]
 
 
 def _rulebook_rows(rows) -> list[dict]:
@@ -982,42 +996,53 @@ def _rule_rows(rows) -> list[dict]:
 # ---------------------------------------------------------------------------
 
 async def export_full_backup() -> dict:
-    """Export all data as a version-5 JSON backup."""
+    """Export all data as a full backup payload.
+
+    Every query is ORDERED, by primary key unless a restore needs another
+    order. Postgres returns an unordered select in heap order, and an UPDATE
+    moves a row to the end of the heap: an edited note would jump to the
+    bottom of `notes`, shifting every byte after its old place. The scheduled
+    backup file (services/backup_file.py) is deduplicated by restic between
+    runs, which only works when unchanged rows serialise to the same bytes in
+    the same place.
+    """
     async with async_session() as session:
-        users = (await session.execute(select(User))).scalars().all()
-        projects = (await session.execute(select(Project))).scalars().all()
-        milestones = (await session.execute(select(Milestone))).scalars().all()
-        notes = (await session.execute(select(Note))).scalars().all()
-        task_logs = (await session.execute(select(TaskLog))).scalars().all()
-        note_drafts = (await session.execute(select(NoteDraft))).scalars().all()
+        users =(await session.execute(select(User).order_by(User.id))).scalars().all()
+        projects = (await session.execute(select(Project).order_by(Project.id))).scalars().all()
+        milestones = (await session.execute(select(Milestone).order_by(Milestone.id))).scalars().all()
+        notes = (await session.execute(select(Note).order_by(Note.id))).scalars().all()
+        task_logs = (await session.execute(select(TaskLog).order_by(TaskLog.id))).scalars().all()
+        note_drafts = (await session.execute(select(NoteDraft).order_by(NoteDraft.id))).scalars().all()
         note_versions = (await session.execute(
             select(NoteVersion).order_by(NoteVersion.note_id, NoteVersion.id)
         )).scalars().all()
         rule_versions = (await session.execute(
             select(RuleVersion).order_by(RuleVersion.rule_id, RuleVersion.id)
         )).scalars().all()
-        settings = (await session.execute(select(Setting))).scalars().all()
-        systems = (await session.execute(select(System))).scalars().all()
+        settings = (await session.execute(select(Setting).order_by(Setting.user_id, Setting.key))).scalars().all()
+        systems = (await session.execute(select(System).order_by(System.id))).scalars().all()
         canonical_systems = (await session.execute(
             select(CanonicalSystem).where(CanonicalSystem.deleted_at.is_(None))
-            .order_by(CanonicalSystem.order_index)
+            .order_by(CanonicalSystem.order_index, CanonicalSystem.id)
         )).scalars().all()
         rule_system_rows = (await session.execute(
             select(rule_systems_t.c.rule_id, CanonicalSystem.slug)
             .join(CanonicalSystem, CanonicalSystem.id == rule_systems_t.c.canonical_id)
+            .order_by(rule_systems_t.c.rule_id, CanonicalSystem.slug)
         )).all()
         rule_moment_rows = (await session.execute(
             select(rule_moments_t.c.rule_id, rule_moments_t.c.moment)
+            .order_by(rule_moments_t.c.rule_id, rule_moments_t.c.moment)
         )).all()
         rule_moment_judgments = (await session.execute(
-            select(RuleMomentJudgment)
+            select(RuleMomentJudgment).order_by(RuleMomentJudgment.id)
         )).scalars().all()
-        rule_relations = (await session.execute(select(RuleRelation))).scalars().all()
-        lesson_rule_links = (await session.execute(select(LessonRuleLink))).scalars().all()
-        lesson_no_rule = (await session.execute(select(LessonNoRule))).scalars().all()
-        record_systems = (await session.execute(select(RecordSystem))).scalars().all()
+        rule_relations = (await session.execute(select(RuleRelation).order_by(RuleRelation.id))).scalars().all()
+        lesson_rule_links = (await session.execute(select(LessonRuleLink).order_by(LessonRuleLink.id))).scalars().all()
+        lesson_no_rule = (await session.execute(select(LessonNoRule).order_by(LessonNoRule.lesson_id))).scalars().all()
+        record_systems = (await session.execute(select(RecordSystem).order_by(RecordSystem.id))).scalars().all()
         supersessions = (
-            await session.execute(select(NoteSupersession))
+            await session.execute(select(NoteSupersession).order_by(NoteSupersession.id))
         ).scalars().all()
         # Parent-first, so a restore can resolve parent_id as it goes rather
         # than needing a second pass — the self-FK is the only ordering
@@ -1026,13 +1051,13 @@ async def export_full_backup() -> dict:
             select(DesignSystem).order_by(DesignSystem.parent_id.nullsfirst(),
                                           DesignSystem.id)
         )).scalars().all()
-        design_tokens = (await session.execute(select(DesignToken))).scalars().all()
-        usage_events = (await session.execute(select(NoteUsageEvent))).scalars().all()
+        design_tokens = (await session.execute(select(DesignToken).order_by(DesignToken.id))).scalars().all()
+        usage_events = (await session.execute(select(NoteUsageEvent).order_by(NoteUsageEvent.id))).scalars().all()
         rule_usage_events = (
-            await session.execute(select(RuleUsageEvent))
+            await session.execute(select(RuleUsageEvent).order_by(RuleUsageEvent.id))
         ).scalars().all()
         system_usage_events = (
-            await session.execute(select(SystemUsageEvent))
+            await session.execute(select(SystemUsageEvent).order_by(SystemUsageEvent.id))
         ).scalars().all()
         # Oldest first, so a restored history reads in the order the dials
         # actually moved — the sequence IS the argument when a surface has been
@@ -1043,30 +1068,33 @@ async def export_full_backup() -> dict:
         moment_mappings = (await session.execute(
             select(MomentMapping).order_by(MomentMapping.id)
         )).scalars().all()
-        repo_bindings = (await session.execute(select(RepoBinding))).scalars().all()
-        code_shapes = (await session.execute(select(CodeShape))).scalars().all()
+        repo_bindings = (await session.execute(select(RepoBinding).order_by(RepoBinding.id))).scalars().all()
+        code_shapes = (await session.execute(select(CodeShape).order_by(CodeShape.id))).scalars().all()
         code_shape_events = (await session.execute(
             select(CodeShapeEvent).order_by(CodeShapeEvent.at, CodeShapeEvent.id)
         )).scalars().all()
         code_shape_uses = (await session.execute(
-            select(CodeShapeUse).order_by(CodeShapeUse.shape_id, CodeShapeUse.snippet_id)
+            select(CodeShapeUse).order_by(CodeShapeUse.shape_id, CodeShapeUse.snippet_id, CodeShapeUse.id)
         )).scalars().all()
-        rulebooks = (await session.execute(select(Rulebook))).scalars().all()
-        topics = (await session.execute(select(RulebookTopic))).scalars().all()
-        rules = (await session.execute(select(Rule))).scalars().all()
+        rulebooks = (await session.execute(select(Rulebook).order_by(Rulebook.id))).scalars().all()
+        topics = (await session.execute(select(RulebookTopic).order_by(RulebookTopic.id))).scalars().all()
+        rules = (await session.execute(select(Rule).order_by(Rule.id))).scalars().all()
         platforms = (await session.execute(
             select(Platform).where(Platform.deleted_at.is_(None))
-            .order_by(Platform.order_index)
+            .order_by(Platform.order_index, Platform.id)
         )).scalars().all()
-        project_platforms = (await session.execute(select(ProjectPlatform))).scalars().all()
-        family_ideas = (await session.execute(select(FamilyIdea))).scalars().all()
+        project_platforms = (await session.execute(select(ProjectPlatform).order_by(
+            ProjectPlatform.project_id, ProjectPlatform.platform_id))).scalars().all()
+        family_ideas = (await session.execute(select(FamilyIdea).order_by(FamilyIdea.note_id))).scalars().all()
         family_idea_platforms = (await session.execute(
-            select(FamilyIdeaPlatform)
+            select(FamilyIdeaPlatform).order_by(
+                FamilyIdeaPlatform.note_id, FamilyIdeaPlatform.platform_id)
         )).scalars().all()
         family_idea_references = (await session.execute(
-            select(FamilyIdeaReference)
+            select(FamilyIdeaReference).order_by(
+                FamilyIdeaReference.idea_id, FamilyIdeaReference.snippet_id)
         )).scalars().all()
-        family_adoptions = (await session.execute(select(FamilyAdoption))).scalars().all()
+        family_adoptions = (await session.execute(select(FamilyAdoption).order_by(FamilyAdoption.id))).scalars().all()
         # Oldest first: a precedent is always an earlier decision, so a restore
         # in this order has every precedent mapped before anything cites it.
         family_decisions = (await session.execute(

src/scribe/services/backup_destination.py

@@ -0,0 +1,261 @@
+"""Where the scheduled backup goes, and getting a backup file there.
+
+Two kinds of destination, both instance-wide and set by an admin:
+
+- **folder**: a directory inside the container, usually a mounted share. A
+  backup service that watches the share (or a person) takes it from there.
+- **s3**: any S3-compatible bucket, addressed by path. Scribe only puts and
+  lists, so an append-only key is enough.
+
+Either way Scribe ADDS one dated file per run and never removes one: keeping
+and thinning the history is the destination's job (services/backup_file.py
+says why the file is shaped the way it is for that).
+
+The settings live as admin settings: `backup_config` holds everything but the
+S3 secret as JSON, and `backup_s3_secret` holds the secret sealed under the
+instance's SECRET_KEY (services/secrets.py). The secret is also kept out of
+every backup (backup._SETTINGS_NOT_EXPORTED), so a backup copied off-site
+does not carry the key that writes to that very place.
+"""
+from __future__ import annotations
+
+import asyncio
+import json
+import os
+import tempfile
+from dataclasses import asdict, dataclass, fields, replace
+from datetime import datetime
+from pathlib import Path
+
+from scribe.services import backup_file
+from scribe.services.backup_s3 import S3Error, S3Target, list_objects, put_file
+from scribe.services.secrets import seal, unseal
+from scribe.services.settings import get_admin_setting, set_admin_setting
+
+CONFIG_SETTING = "backup_config"
+SECRET_SETTING = "backup_s3_secret"
+DESTINATION_KINDS = ("folder", "s3")
+DEFAULT_FOLDER = "/backups"
+# 02:00 UTC: before the 03:30 trash purge and the 04:00 DB maintenance, so the
+# backup never contends with a VACUUM, and the night's deletions are still
+# recoverable from the backup taken just before them.
+DEFAULT_HOUR = 2
+
+
+class DestinationError(Exception):
+    """Why the destination can't take a backup, said to the person."""
+
+
+@dataclass(frozen=True)
+class BackupConfig:
+    enabled: bool = False
+    hour: int = DEFAULT_HOUR
+    destination: str = "folder"
+    folder_path: str = DEFAULT_FOLDER
+    s3_endpoint: str = ""
+    s3_bucket: str = ""
+    s3_access_key: str = ""
+    s3_region: str = "us-east-1"
+    s3_verify_tls: bool = True
+
+    @classmethod
+    def from_dict(cls, raw: dict) -> "BackupConfig":
+        """Lenient: unknown keys are ignored and bad values fall back to the
+        default, so an edited or older stored config still loads."""
+        known = {f.name: f for f in fields(cls)}
+        values = {}
+        for name, value in (raw or {}).items():
+            f = known.get(name)
+            if f is None:
+                continue
+            default = f.default
+            if isinstance(default, bool):
+                values[name] = value if isinstance(value, bool) else default
+            elif isinstance(default, int):
+                try:
+                    values[name] = int(value)
+                except (TypeError, ValueError):
+                    values[name] = default
+            else:
+                values[name] = str(value).strip() if value is not None else default
+        config = cls(**values)
+        if not 0 <= config.hour <= 23:
+            config = replace(config, hour=DEFAULT_HOUR)
+        if config.destination not in DESTINATION_KINDS:
+            config = replace(config, destination="folder")
+        return config
+
+    def to_dict(self) -> dict:
+        return asdict(self)
+
+    def problems(self, *, has_secret: bool) -> list[str]:
+        """What stops this config from sending a backup, in words. Empty when
+        it is complete (which says nothing about whether the place answers —
+        check_destination does that)."""
+        out: list[str] = []
+        if self.destination == "folder":
+            if not self.folder_path.startswith("/"):
+                out.append("The folder must be an absolute path inside the container.")
+        else:
+            if not self.s3_endpoint.startswith(("http://", "https://")):
+                out.append("The S3 address must start with http:// or https://.")
+            if not self.s3_bucket:
+                out.append("The bucket name is missing.")
+            if not self.s3_access_key:
+                out.append("The access key is missing.")
+            if not has_secret:
+                out.append("The secret key is missing.")
+        return out
+
+
+def apply_changes(current: BackupConfig, data: dict) -> BackupConfig:
+    """`current` with the fields an admin sent changed, checked strictly.
+
+    `from_dict` forgives a stored value it can't read; a value a person just
+    typed is refused instead, with the reason, so a typo in the hour does not
+    silently become 02:00. Keys that aren't config fields (the secret, say)
+    are ignored here.
+    """
+    names = {f.name for f in fields(BackupConfig)}
+    changes = {k: v for k, v in (data or {}).items() if k in names}
+    if "hour" in changes:
+        try:
+            hour = int(changes["hour"])
+        except (TypeError, ValueError):
+            raise ValueError("The hour must be a whole number from 0 to 23.") from None
+        if not 0 <= hour <= 23:
+            raise ValueError("The hour must be from 0 to 23 (UTC).")
+        changes["hour"] = hour
+    if "destination" in changes and changes["destination"] not in DESTINATION_KINDS:
+        raise ValueError("The destination must be a folder or an S3 bucket.")
+    for name in ("enabled", "s3_verify_tls"):
+        if name in changes and not isinstance(changes[name], bool):
+            raise ValueError(f"{name} must be true or false.")
+    return BackupConfig.from_dict({**current.to_dict(), **changes})
+
+
+async def load_config() -> BackupConfig:
+    raw = await get_admin_setting(CONFIG_SETTING, "")
+    try:
+        data = json.loads(raw) if raw else {}
+    except ValueError:
+        data = {}
+    return BackupConfig.from_dict(data if isinstance(data, dict) else {})
+
+
+async def save_config(config: BackupConfig) -> None:
+    await set_admin_setting(CONFIG_SETTING, json.dumps(config.to_dict(), sort_keys=True))
+
+
+async def load_secret() -> tuple[str | None, bool]:
+    """(the S3 secret or None, whether one is stored).
+
+    Stored but None means it was sealed under another SECRET_KEY — after a
+    restore onto a different instance, say — and has to be entered again.
+    """
+    stored = await get_admin_setting(SECRET_SETTING, "")
+    return (unseal(stored), bool(stored))
+
+
+async def save_secret(secret: str) -> None:
+    await set_admin_setting(SECRET_SETTING, seal(secret) if secret else "")
+
+
+def s3_target(config: BackupConfig, secret: str) -> S3Target:
+    return S3Target(
+        endpoint=config.s3_endpoint, bucket=config.s3_bucket,
+        access_key=config.s3_access_key, secret_key=secret,
+        region=config.s3_region or "us-east-1", verify_tls=config.s3_verify_tls,
+    )
+
+
+@dataclass(frozen=True)
+class Delivered:
+    name: str
+    size_bytes: int
+    where: str
+
+
+async def deliver(
+    payload: dict, config: BackupConfig, secret: str | None, at: datetime,
+    *, app_version: str, transport=None,
+) -> Delivered:
+    """Write `payload` as the dated backup file and put it at the destination.
+
+    The file is written off the event loop: serialising a full export is
+    seconds of CPU on a large corpus, and the app keeps serving meanwhile.
+    """
+    problems = config.problems(has_secret=bool(secret))
+    if problems:
+        raise DestinationError(" ".join(problems))
+
+    if config.destination == "folder":
+        folder = Path(config.folder_path)
+        if not folder.is_dir():
+            raise DestinationError(
+                f"{folder} isn't there. Mount the backup share at that path, "
+                "or choose the folder it is mounted at."
+            )
+        try:
+            path = await asyncio.to_thread(
+                backup_file.write_archive, payload, folder, at, app_version=app_version,
+            )
+        except PermissionError:
+            raise DestinationError(f"Scribe isn't allowed to write in {folder}.") from None
+        return Delivered(path.name, path.stat().st_size, str(folder))
+
+    target = s3_target(config, secret or "")
+    with tempfile.TemporaryDirectory(prefix="scribe-backup-") as staging:
+        path = await asyncio.to_thread(
+            backup_file.write_archive, payload, Path(staging), at, app_version=app_version,
+        )
+        size = path.stat().st_size
+        try:
+            await put_file(target, path.name, path, transport=transport)
+        except S3Error as exc:
+            raise DestinationError(str(exc)) from None
+    return Delivered(path.name, size, target.url())
+
+
+async def check_destination(
+    config: BackupConfig, secret: str | None, *, transport=None,
+) -> str:
+    """Ask the destination whether it would take a backup, without sending
+    one. Returns what was found, in words; raises DestinationError otherwise.
+
+    The folder check writes and removes a dot-file, which a folder watcher
+    ignores. The S3 check only lists: an append-only key could add a test
+    object and then never remove it.
+    """
+    problems = config.problems(has_secret=bool(secret))
+    if problems:
+        raise DestinationError(" ".join(problems))
+
+    if config.destination == "folder":
+        folder = Path(config.folder_path)
+        if not folder.is_dir():
+            raise DestinationError(f"{folder} isn't there, or isn't a folder.")
+        probe = folder / ".scribe-write-check"
+        try:
+            await asyncio.to_thread(probe.write_bytes, b"")
+            await asyncio.to_thread(probe.unlink)
+        except PermissionError:
+            raise DestinationError(f"Scribe isn't allowed to write in {folder}.") from None
+        except OSError as exc:
+            raise DestinationError(f"Writing in {folder} failed: {exc.strerror or exc}.") from None
+        count = await asyncio.to_thread(_count_backups, folder)
+        return f"{folder} is writable and holds {count} Scribe backup(s)."
+
+    target = s3_target(config, secret or "")
+    try:
+        objects = await list_objects(target, backup_file.NAME_PREFIX, transport=transport)
+    except S3Error as exc:
+        raise DestinationError(str(exc)) from None
+    return f"Bucket {config.s3_bucket} answered and holds {len(objects)} Scribe backup(s)."
+
+
+def _count_backups(folder: Path) -> int:
+    return sum(
+        1 for name in os.listdir(folder)
+        if name.startswith(backup_file.NAME_PREFIX) and name.endswith(backup_file.NAME_SUFFIX)
+    )

src/scribe/services/backup_file.py

@@ -0,0 +1,214 @@
+"""The full backup as a file on disk: one uncompressed tar per run.
+
+The scheduled backup lands somewhere that keeps a history of it — a folder a
+backup service watches, an S3 bucket — and from there it is usually copied
+off-site by a deduplicating tool such as restic. Everything about the layout
+serves that copy: restic cuts a file into chunks at boundaries chosen by the
+content, so a chunk it has seen before costs nothing to store again. The file
+is therefore laid out so that what did not change between two runs is the
+same bytes, in the same place, both times.
+
+- **Never compressed.** Compression rewrites every byte after the first
+  change, so no chunk would ever match. restic compresses its own repository
+  after deduplicating.
+- **One member per payload section**, `<section>.jsonl`, one row per line.
+  Rows arrive in primary-key order (export_full_backup orders every query)
+  and each is written with sorted keys and no spacing, so an unchanged row is
+  identical bytes, an edit changes one line, and rows appended to a growing
+  table land at the end of its member. A busy table never shifts the bytes of
+  a quiet one.
+- **The parts that change every run live in `manifest.json` only**: the
+  export time, the app version and the row counts. The member headers are
+  fixed (mtime 0, owner 0, mode 0644) and the members come in one order: the
+  manifest, then the sections by name.
+- **One file per run, under a new dated name**, so a service that keeps one
+  file per backup can thin the history on its own schedule.
+
+The payload inside is exactly `export_full_backup()`'s, so the coverage
+guards on that payload (tests/test_services_backup.py) cover this file too,
+and `read_archive` hands `restore_full_backup` the same dict it always took.
+"""
+from __future__ import annotations
+
+import io
+import json
+import os
+import tarfile
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import IO
+
+ARCHIVE_FORMAT = "scribe-backup-tar"
+ARCHIVE_FORMAT_VERSION = 1
+MANIFEST_NAME = "manifest.json"
+SECTION_SUFFIX = ".jsonl"
+NAME_PREFIX = "scribe-"
+NAME_SUFFIX = ".tar"
+
+
+def archive_name(at: datetime) -> str:
+    """`scribe-20261015T030000Z.tar` — UTC, sortable, one per run."""
+    return f"{NAME_PREFIX}{at.astimezone(timezone.utc):%Y%m%dT%H%M%SZ}{NAME_SUFFIX}"
+
+
+def _encode(value) -> str:
+    # default=str matches the admin download's json.dumps, so a datetime that
+    # reaches the payload unconverted reads back the same way on restore.
+    return json.dumps(
+        value, sort_keys=True, separators=(",", ":"), ensure_ascii=False, default=str,
+    )
+
+
+def _split(payload: dict) -> tuple[dict, dict[str, list]]:
+    """Separate the row sections from the header (version, scope, notices).
+
+    A section is a list of rows under a plain name. `_not_included` is a list
+    too, but of table names: it is a notice, so it stays in the header.
+    """
+    header: dict = {}
+    sections: dict[str, list] = {}
+    for key, value in payload.items():
+        if isinstance(value, list) and not key.startswith("_"):
+            sections[key] = value
+        else:
+            header[key] = value
+    return header, sections
+
+
+def _member(name: str, data: bytes) -> tuple[tarfile.TarInfo, io.BytesIO]:
+    info = tarfile.TarInfo(name)
+    info.size = len(data)
+    info.mtime = 0
+    info.mode = 0o644
+    info.uid = info.gid = 0
+    info.uname = info.gname = ""
+    return info, io.BytesIO(data)
+
+
+def write_archive_to(payload: dict, out: IO[bytes], *, app_version: str) -> dict:
+    """Write `payload` as the backup tar to an open binary file.
+
+    Returns the manifest, so the caller can report row counts without
+    reading the file back.
+    """
+    header, sections = _split(payload)
+    names = sorted(sections)
+    manifest = {
+        "format": ARCHIVE_FORMAT,
+        "format_version": ARCHIVE_FORMAT_VERSION,
+        "app_version": app_version,
+        "header": header,
+        "sections": {name: len(sections[name]) for name in names},
+    }
+    with tarfile.open(fileobj=out, mode="w", format=tarfile.USTAR_FORMAT) as tar:
+        tar.addfile(*_member(
+            MANIFEST_NAME, (json.dumps(manifest, sort_keys=True, indent=2,
+                                       default=str) + "\n").encode(),
+        ))
+        for name in names:
+            data = b"".join(
+                _encode(row).encode() + b"\n" for row in sections[name]
+            )
+            tar.addfile(*_member(name + SECTION_SUFFIX, data))
+    return manifest
+
+
+def write_archive(
+    payload: dict, directory: Path, at: datetime, *, app_version: str,
+) -> Path:
+    """Write the backup tar into `directory` under its dated name, atomically.
+
+    The bytes go to a dot-prefixed `.part` file first, are fsync'd, and only
+    then renamed to the final name. A watcher of the folder (or anything else
+    listing it) never sees half a backup under a backup's name, and a run that
+    fails leaves nothing behind but the error.
+    """
+    directory.mkdir(parents=True, exist_ok=True)
+    final = directory / archive_name(at)
+    part = directory / f".{final.name}.part"
+    try:
+        with open(part, "wb") as f:
+            write_archive_to(payload, f, app_version=app_version)
+            f.flush()
+            os.fsync(f.fileno())
+        os.replace(part, final)
+    except BaseException:
+        part.unlink(missing_ok=True)
+        raise
+    _fsync_dir(directory)
+    return final
+
+
+def _fsync_dir(directory: Path) -> None:
+    """Make the rename itself durable. Some filesystems (and NFS mounts)
+    refuse a directory fsync; the file's own fsync already happened, so that
+    refusal is not a failure of the backup."""
+    try:
+        fd = os.open(directory, os.O_RDONLY)
+    except OSError:
+        return
+    try:
+        os.fsync(fd)
+    except OSError:
+        pass
+    finally:
+        os.close(fd)
+
+
+class ArchiveError(ValueError):
+    """The file is not a backup tar this version can read, or is damaged."""
+
+
+def is_archive(fileobj: IO[bytes]) -> bool:
+    """True when `fileobj` (seekable, at its start) holds a tar. Leaves the
+    position where it was, so the caller can fall back to reading JSON."""
+    pos = fileobj.tell()
+    try:
+        head = fileobj.read(tarfile.BLOCKSIZE)
+    finally:
+        fileobj.seek(pos)
+    return len(head) == tarfile.BLOCKSIZE and head[257:262] == b"ustar"
+
+
+def read_archive(fileobj: IO[bytes]) -> dict:
+    """Read a backup tar back into the payload `restore_full_backup` takes.
+
+    Every section the manifest names must be present with the row count it
+    names: a member that went missing or was cut short is an error here, not
+    a restore that reports success with less than it was given.
+    """
+    try:
+        tar = tarfile.open(fileobj=fileobj, mode="r:")
+    except tarfile.TarError as exc:
+        raise ArchiveError(f"not a backup archive: {exc}") from exc
+    with tar:
+        manifest = json.loads(_read_member(tar, MANIFEST_NAME))
+        if manifest.get("format") != ARCHIVE_FORMAT:
+            raise ArchiveError("not a Scribe backup archive (no Scribe manifest)")
+        if manifest.get("format_version", 0) > ARCHIVE_FORMAT_VERSION:
+            raise ArchiveError(
+                f"archive format {manifest['format_version']} is newer than "
+                f"this Scribe reads ({ARCHIVE_FORMAT_VERSION}); upgrade first"
+            )
+        payload = dict(manifest.get("header") or {})
+        for name, count in (manifest.get("sections") or {}).items():
+            raw = _read_member(tar, name + SECTION_SUFFIX)
+            rows = [json.loads(line) for line in raw.splitlines() if line]
+            if len(rows) != count:
+                raise ArchiveError(
+                    f"{name}{SECTION_SUFFIX} holds {len(rows)} rows; "
+                    f"the manifest says {count}"
+                )
+            payload[name] = rows
+    return payload
+
+
+def _read_member(tar: tarfile.TarFile, name: str) -> bytes:
+    try:
+        member = tar.getmember(name)
+    except KeyError as exc:
+        raise ArchiveError(f"the archive has no {name}") from exc
+    f = tar.extractfile(member)
+    if f is None:
+        raise ArchiveError(f"{name} in the archive is not a file")
+    return f.read()

src/scribe/services/backup_job.py

@@ -0,0 +1,188 @@
+"""The scheduled backup: when it runs, running it, and what each run did.
+
+ONE APScheduler job (services/scheduler.py) fires at the top of every hour and
+reads the config at fire time: when backups are on and the hour is the chosen
+one, it runs. Reading at fire time rather than scheduling the chosen hour
+keeps the app's startup free of a database read (#4181 — a lifespan hook that
+waits on the database is a worker that never serves), and a changed hour
+needs no reschedule.
+
+A run exports the full backup, writes it as the dated tar
+(services/backup_file.py) and delivers it (services/backup_destination.py).
+Its outcome is kept as one entry in a bounded list — the newest
+`KEEP_RUNS` — in the `backup_runs` admin setting, like DB maintenance keeps
+its last run. A failure is logged as an error AND kept with its reason in
+words, so the Settings page can say what went wrong without anyone reading
+logs.
+"""
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+import os
+import time
+from datetime import datetime, timedelta, timezone
+
+from apscheduler.triggers.cron import CronTrigger
+
+from scribe.services.backup_destination import (
+    DestinationError, deliver, load_config, load_secret,
+)
+from scribe.services.scheduler import ScheduledJob
+from scribe.services.settings import get_admin_setting, set_admin_setting
+
+logger = logging.getLogger(__name__)
+
+RUNS_SETTING = "backup_runs"
+KEEP_RUNS = 30
+# A daily backup that hasn't succeeded for this long is overdue: a day plus
+# room for one slow run, so a backup that lands at 02:40 instead of 02:05
+# does not flicker the page into a warning.
+STALE_AFTER = timedelta(hours=26)
+# The whole run's deadline (rule 156: a wait with no deadline is a bug). The
+# S3 upload has its own 15-minute timeout, but a folder on a hung NFS mount
+# has none, and a run that never returns holds the lock — every later backup
+# would be skipped as "already running", silently, forever.
+RUN_DEADLINE = timedelta(minutes=60)
+
+_LOCK = asyncio.Lock()
+_TASKS: set[asyncio.Task] = set()
+
+
+class BackupBusy(Exception):
+    """A backup is already running."""
+
+
+def is_running() -> bool:
+    return _LOCK.locked()
+
+
+async def list_runs() -> list[dict]:
+    """The kept runs, newest first."""
+    raw = await get_admin_setting(RUNS_SETTING, "")
+    try:
+        runs = json.loads(raw) if raw else []
+    except ValueError:
+        return []
+    return [r for r in runs if isinstance(r, dict)] if isinstance(runs, list) else []
+
+
+async def _record(run: dict) -> None:
+    runs = [run, *await list_runs()][:KEEP_RUNS]
+    await set_admin_setting(RUNS_SETTING, json.dumps(runs))
+
+
+def summarise(runs: list[dict], *, enabled: bool, now: datetime) -> dict:
+    """What the page leads with: when the last good backup was, and whether
+    backups are overdue. Pure, so the stale rule is tested on its own."""
+    last_ok = next((r for r in runs if r.get("status") == "ok"), None)
+    last_ok_at = _parse(last_ok.get("finished_at")) if last_ok else None
+    if not enabled:
+        overdue = False
+    elif last_ok_at is None:
+        # On, and never succeeded: overdue once the first scheduled chance
+        # has passed. Before any run at all, there is nothing to be late for.
+        overdue = bool(runs)
+    else:
+        overdue = now - last_ok_at > STALE_AFTER
+    return {
+        "last_run": runs[0] if runs else None,
+        "last_success": last_ok,
+        "overdue": overdue,
+    }
+
+
+def _parse(value) -> datetime | None:
+    try:
+        return datetime.fromisoformat(value)
+    except (TypeError, ValueError):
+        return None
+
+
+async def run_backup(trigger: str) -> dict:
+    """Run one backup now and keep its outcome. Raises BackupBusy when one is
+    already running; never raises for a failed backup — that is the outcome.
+    """
+    if _LOCK.locked():
+        raise BackupBusy()
+    async with _LOCK:
+        started = datetime.now(timezone.utc)
+        clock = time.monotonic()
+        run: dict = {"started_at": started.isoformat(), "trigger": trigger}
+        try:
+            config = await load_config()
+            secret, _ = await load_secret()
+            run["destination"] = config.destination
+            delivered = await asyncio.wait_for(
+                _export_and_deliver(config, secret, started), RUN_DEADLINE.total_seconds(),
+            )
+            run.update(status="ok", file=delivered.name,
+                       size_bytes=delivered.size_bytes, where=delivered.where)
+        except (TimeoutError, asyncio.TimeoutError):
+            run.update(status="failed", error=(
+                f"The backup didn't finish within {int(RUN_DEADLINE.total_seconds() // 60)} "
+                "minutes and was abandoned. The destination may be hanging."
+            ))
+        except DestinationError as exc:
+            run.update(status="failed", error=str(exc))
+        except Exception as exc:
+            logger.exception("backup run failed")
+            run.update(status="failed", error=f"The backup failed: {exc}")
+        run["finished_at"] = datetime.now(timezone.utc).isoformat()
+        run["duration_s"] = round(time.monotonic() - clock, 1)
+        if run["status"] == "ok":
+            logger.info("backup: wrote %s (%d bytes) to %s in %.1fs",
+                        run["file"], run["size_bytes"], run["where"], run["duration_s"])
+        else:
+            logger.error("backup: %s", run["error"])
+        try:
+            await _record(run)
+        except Exception:
+            # The backup itself is done (or failed) either way; losing its
+            # entry in the history must not turn into an unhandled error in a
+            # scheduler thread. The log line above still says what happened.
+            logger.exception("backup: could not keep this run in the history")
+        return run
+
+
+async def _export_and_deliver(config, secret, started):
+    from scribe.services.backup import export_full_backup
+    payload = await export_full_backup()
+    return await deliver(
+        payload, config, secret, started,
+        app_version=os.environ.get("APP_VERSION", "dev"),
+    )
+
+
+def start_backup_now() -> None:
+    """Start a manual backup in the background (the request returns at once;
+    the page reads progress from `is_running` and the run list). Raises
+    BackupBusy rather than queueing a second run behind the first."""
+    if _LOCK.locked():
+        raise BackupBusy()
+    task = asyncio.get_running_loop().create_task(run_backup("manual"))
+    _TASKS.add(task)
+    task.add_done_callback(_TASKS.discard)
+
+
+async def _hourly_check() -> None:
+    config = await load_config()
+    if not config.enabled or datetime.now(timezone.utc).hour != config.hour:
+        return
+    try:
+        await run_backup("scheduled")
+    except BackupBusy:
+        logger.info("backup: a run is already going; the scheduled one is skipped")
+
+
+_JOB = ScheduledJob("backup_hourly_check", _hourly_check, label="Backup")
+
+
+def start_backup_scheduler(loop: asyncio.AbstractEventLoop) -> None:
+    _JOB.start(loop, CronTrigger(minute=0, timezone="UTC"),
+               describe="hourly; runs at the configured UTC hour")
+
+
+def stop_backup_scheduler() -> None:
+    _JOB.stop()

src/scribe/services/backup_s3.py

@@ -0,0 +1,199 @@
+"""A small S3 client for sending backups: AWS Signature Version 4 over httpx,
+path-style addressing (endpoint/bucket/key), and the two calls a backup needs.
+
+Ported from FabledSteward's `steward/backup/s3.py`, which is tested against
+the same AWS examples as tests/test_services_backup_s3.py.
+
+Scribe puts and lists; it never deletes. The bucket keeps the history and
+thins it on its own policy, so a key that may only add objects is enough —
+and is the safer key to hand a service, because a stolen one cannot wipe the
+history it writes into.
+
+Written by hand rather than with boto3: two calls don't justify a large
+dependency, and signing is a pure function tests can hold to AWS's own
+published examples.
+"""
+from __future__ import annotations
+
+import asyncio
+import hashlib
+import hmac
+import xml.etree.ElementTree as ET
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from pathlib import Path
+from urllib.parse import quote
+
+import httpx
+
+EMPTY_SHA256 = hashlib.sha256(b"").hexdigest()
+TIMEOUT_SECONDS = 30.0
+# A backup is tens of megabytes; a slow link still gets this long to send one.
+UPLOAD_TIMEOUT_SECONDS = 15 * 60
+CHUNK = 1 << 20
+
+
+class S3Error(Exception):
+    """Why a call to the bucket failed, said to the person."""
+
+
+@dataclass(frozen=True)
+class S3Target:
+    endpoint: str          # http(s)://host[:port][/prefix], no bucket
+    bucket: str
+    access_key: str
+    secret_key: str
+    region: str = "us-east-1"
+    verify_tls: bool = True
+
+    def url(self, key: str = "") -> str:
+        base = f"{self.endpoint.rstrip('/')}/{self.bucket}"
+        return f"{base}/{key}" if key else f"{base}/"
+
+
+@dataclass(frozen=True)
+class StoredObject:
+    key: str
+    size: int
+    last_modified: str
+
+
+# ── Signing ──────────────────────────────────────────────────────────────────
+
+def _encode(text: str, keep_slash: bool = False) -> str:
+    return quote(text, safe="-_.~" + ("/" if keep_slash else ""))
+
+
+def _hmac(key: bytes, msg: str) -> bytes:
+    return hmac.new(key, msg.encode(), hashlib.sha256).digest()
+
+
+def sign(method: str, url: str, headers: dict[str, str], payload_sha256: str,
+         access_key: str, secret_key: str, region: str, at: datetime,
+         service: str = "s3") -> dict[str, str]:
+    """The request's headers with x-amz-date, x-amz-content-sha256 and
+    Authorization added. Every header given is signed, Host always."""
+    u = httpx.URL(url)
+    stamp = at.astimezone(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
+    day = stamp[:8]
+    out = dict(headers)
+    out["x-amz-date"] = stamp
+    out["x-amz-content-sha256"] = payload_sha256
+    signed = {k.lower(): " ".join(str(v).split()) for k, v in out.items()}
+    signed["host"] = u.netloc.decode()
+    names = sorted(signed)
+
+    query = sorted((_encode(k), _encode(v)) for k, v in u.params.multi_items())
+    canonical = "\n".join([
+        method.upper(),
+        _encode(u.path or "/", keep_slash=True),
+        "&".join(f"{k}={v}" for k, v in query),
+        "".join(f"{n}:{signed[n]}\n" for n in names),
+        ";".join(names),
+        payload_sha256,
+    ])
+    scope = f"{day}/{region}/{service}/aws4_request"
+    to_sign = "\n".join(["AWS4-HMAC-SHA256", stamp, scope,
+                         hashlib.sha256(canonical.encode()).hexdigest()])
+    key = _hmac(_hmac(_hmac(_hmac(f"AWS4{secret_key}".encode(), day), region),
+                      service), "aws4_request")
+    signature = hmac.new(key, to_sign.encode(), hashlib.sha256).hexdigest()
+    out["Authorization"] = (f"AWS4-HMAC-SHA256 Credential={access_key}/{scope}, "
+                            f"SignedHeaders={';'.join(names)}, Signature={signature}")
+    return out
+
+
+# ── Calls ────────────────────────────────────────────────────────────────────
+
+def _says(response: httpx.Response) -> str:
+    """An S3 error response in words: its Code and Message when it has them."""
+    try:
+        root = ET.fromstring(response.content)
+        code = root.findtext("Code") or ""
+        message = root.findtext("Message") or ""
+    except ET.ParseError:
+        code, message = "", ""
+    hints = {
+        "SignatureDoesNotMatch": "the secret key is wrong",
+        "InvalidAccessKeyId": "the access key isn't known there",
+        "NoSuchBucket": "there is no such bucket",
+        "AccessDenied": "the key isn't allowed to do that",
+    }
+    said = hints.get(code) or message or response.reason_phrase
+    return f"HTTP {response.status_code}{f' {code}' if code else ''}: {said}."
+
+
+def _reach_error(target: S3Target, exc: httpx.HTTPError) -> S3Error:
+    if isinstance(exc, httpx.TimeoutException):
+        return S3Error(f"{target.endpoint} didn't answer in time.")
+    return S3Error(f"Couldn't reach {target.endpoint}: {exc}.")
+
+
+def file_sha256(path: Path) -> str:
+    h = hashlib.sha256()
+    with open(path, "rb") as f:
+        while chunk := f.read(CHUNK):
+            h.update(chunk)
+    return h.hexdigest()
+
+
+async def put_file(target: S3Target, key: str, path: Path, *,
+                   transport: httpx.AsyncBaseTransport | None = None,
+                   now: datetime | None = None) -> None:
+    """Upload one file as one object, streamed from disk."""
+    digest = await asyncio.to_thread(file_sha256, path)
+    size = path.stat().st_size
+    url = target.url(key)
+    headers = sign("PUT", url, {"Content-Length": str(size),
+                                "Content-Type": "application/x-tar"},
+                   digest, target.access_key, target.secret_key, target.region,
+                   now or datetime.now(timezone.utc))
+
+    async def body():
+        with open(path, "rb") as f:
+            while chunk := f.read(CHUNK):
+                yield chunk
+
+    try:
+        async with httpx.AsyncClient(verify=target.verify_tls,
+                                     timeout=UPLOAD_TIMEOUT_SECONDS,
+                                     transport=transport) as client:
+            response = await client.put(url, headers=headers, content=body())
+    except httpx.HTTPError as exc:
+        raise _reach_error(target, exc) from None
+    if response.status_code // 100 != 2:
+        raise S3Error(f"Upload of {key} refused: {_says(response)}")
+
+
+async def list_objects(target: S3Target, prefix: str = "", limit: int = 1000, *,
+                       transport: httpx.AsyncBaseTransport | None = None,
+                       now: datetime | None = None) -> list[StoredObject]:
+    """Up to `limit` objects under `prefix` (ListObjectsV2, one page)."""
+    params = {"list-type": "2", "max-keys": str(limit)}
+    if prefix:
+        params["prefix"] = prefix
+    url = str(httpx.URL(target.url(), params=params))
+    headers = sign("GET", url, {}, EMPTY_SHA256, target.access_key, target.secret_key,
+                   target.region, now or datetime.now(timezone.utc))
+    try:
+        async with httpx.AsyncClient(verify=target.verify_tls, timeout=TIMEOUT_SECONDS,
+                                     transport=transport) as client:
+            response = await client.get(url, headers=headers)
+    except httpx.HTTPError as exc:
+        raise _reach_error(target, exc) from None
+    if response.status_code != 200:
+        raise S3Error(f"Listing {target.bucket} refused: {_says(response)}")
+    try:
+        root = ET.fromstring(response.content)
+    except ET.ParseError:
+        raise S3Error(
+            f"{target.endpoint} answered a listing with something that isn't S3's XML."
+        ) from None
+    out = []
+    for item in root.iter():
+        if item.tag.rsplit("}", 1)[-1] != "Contents":
+            continue
+        fields = {child.tag.rsplit("}", 1)[-1]: (child.text or "") for child in item}
+        out.append(StoredObject(fields.get("Key", ""), int(fields.get("Size") or 0),
+                                fields.get("LastModified", "")))
+    return out

src/scribe/services/forge.py

@@ -358,6 +358,22 @@ _FORGE_CLASSES: dict[str, type[ForgeAdapter]] = {
 }
 
 
+def _token_of(row) -> str:
+    """A keyring row's token, opened; empty when this instance can't open it,
+    which build_adapter reads as "this connection cannot serve reads"."""
+    from scribe.services.forge_connections import connection_token
+
+    token = connection_token(row)
+    if token is None:
+        logger.warning(
+            "forge connection %s (%s) holds a token this instance can't open — "
+            "SECRET_KEY changed since it was saved; its owner must enter it again",
+            row.id, row.host,
+        )
+        return ""
+    return token
+
+
 def build_adapter(
     kind: str, base_url: str, token: str, *, transport=None
 ) -> ForgeAdapter | None:
@@ -459,7 +475,7 @@ async def get_forges(
     if pinned_id:
         pin = next((r for r in rows if r.id == pinned_id), None)
         if pin is not None:
-            adapter = build_adapter(pin.kind, pin.base_url, pin.token, transport=transport)
+            adapter = build_adapter(pin.kind, pin.base_url, _token_of(pin), transport=transport)
             return ForgeSelector((adapter,) if adapter is not None else ())
         logger.warning(
             "project %s pins forge connection %s the owner (%s) does not hold — pin ignored",
@@ -468,7 +484,7 @@ async def get_forges(
 
     adapters: list[ForgeAdapter] = []
     for row in rows:
-        adapter = build_adapter(row.kind, row.base_url, row.token, transport=transport)
+        adapter = build_adapter(row.kind, row.base_url, _token_of(row), transport=transport)
         if adapter is not None:
             adapters.append(adapter)
     if role == "admin":

src/scribe/services/forge_connections.py

@@ -4,7 +4,9 @@ A connection is a user's read-only credential for one forge host; one row per
 (user, host) keeps host-keyed resolution deterministic with no default-pointer
 machinery. Everything here is own-rows-only: a connection is a credential, and
 no caller — admin included — reads or edits another user's. The token never
-leaves the server (model.to_dict omits it; routes mask "set/unset").
+leaves the server (model.to_dict omits it; routes mask "set/unset"), and it is
+stored sealed (services/secrets.py): a copy of the database alone cannot read
+it. Every reader opens it through `connection_token`.
 
 Validation matches what build_adapter will accept, checked here so a bad
 value is a 400 at the form instead of a silently dead keyring row.
@@ -17,6 +19,14 @@ from scribe.models import async_session
 from scribe.models.forge_connection import ForgeConnection
 from scribe.models.project import Project
 from scribe.services.forge import FORGE_KINDS, host_of
+from scribe.services.secrets import seal, unseal
+
+
+def connection_token(row: ForgeConnection) -> str | None:
+    """The connection's token in the clear, or None when this instance can't
+    open it — sealed under a different SECRET_KEY (a restore onto a new
+    instance, a rotated key). The owner enters it again; nothing else can."""
+    return unseal(row.token_sealed)
 
 
 def validate_connection(kind: str, base_url: str) -> str | None:
@@ -82,7 +92,8 @@ async def create_connection(
                 "resolution is by host, so a second row could never be reached"
             )
         row = ForgeConnection(
-            user_id=user_id, kind=kind, base_url=base_url, host=host, token=token
+            user_id=user_id, kind=kind, base_url=base_url, host=host,
+            token_sealed=seal(token),
         )
         session.add(row)
         await session.commit()
@@ -135,7 +146,7 @@ async def update_connection(
         row.base_url = new_base
         row.host = new_host
         if token:
-            row.token = token
+            row.token_sealed = seal(token)
         await session.commit()
         await session.refresh(row)
         return row

src/scribe/services/secrets.py

@@ -0,0 +1,44 @@
+"""Sealing a credential Scribe stores so the database alone cannot read it.
+
+Fernet (AES-128-CBC with an HMAC) under a key derived from the instance's
+SECRET_KEY. A copy of the database, or of a backup made from it, then holds
+only ciphertext: reading the credential also needs the SECRET_KEY, which lives
+in the deployment's environment rather than in the data.
+
+The derivation is labelled and versioned, so this key is never the same bytes
+as anything else made from SECRET_KEY (the session-cookie signing uses it
+raw). An instance that changes its SECRET_KEY can no longer open what it
+sealed before: `unseal` answers None, and the caller asks for the credential
+again rather than failing on it.
+"""
+from __future__ import annotations
+
+import base64
+import hashlib
+
+from cryptography.fernet import Fernet, InvalidToken
+
+from scribe.config import Config
+
+_LABEL = b"scribe.secrets.v1\x00"
+
+
+def _fernet() -> Fernet:
+    digest = hashlib.sha256(_LABEL + Config.SECRET_KEY.encode()).digest()
+    return Fernet(base64.urlsafe_b64encode(digest))
+
+
+def seal(plain: str) -> str:
+    """Encrypt `plain` for storage; the result is ASCII."""
+    return _fernet().encrypt(plain.encode()).decode()
+
+
+def unseal(token: str) -> str | None:
+    """Decrypt a value `seal` produced, or None when it can't be opened —
+    sealed under a different SECRET_KEY, or not a sealed value at all."""
+    if not token:
+        return None
+    try:
+        return _fernet().decrypt(token.encode()).decode()
+    except (InvalidToken, ValueError):
+        return None

tests/test_forge_token_sealing.py

@@ -0,0 +1,47 @@
+"""Forge-connection tokens are stored sealed (milestone 523, #5877).
+
+Migration 0122 seals the rows that existed before, with its OWN frozen copy of
+the v1 key derivation (so a later edit to services/secrets.py can never change
+what an old migration writes). That copy is only right while it agrees with
+what the app opens today — which is what this pins. If secrets.py ever moves
+to a v2 key, it must still open v1 values, and this test says so.
+"""
+import importlib.util
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import patch
+
+from scribe.services import secrets
+from scribe.services.forge_connections import connection_token
+
+_MIGRATION = Path(__file__).resolve().parents[1] / "alembic" / "versions" / "0122_seal_forge_tokens.py"
+
+
+def _migration():
+    spec = importlib.util.spec_from_file_location("migration_0122", _MIGRATION)
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+def test_what_the_migration_seals_the_app_opens():
+    sealed = _migration().seal_v1("gitea-read-token")
+    assert "gitea-read-token" not in sealed
+    assert secrets.unseal(sealed) == "gitea-read-token"
+
+
+def test_a_row_reads_through_connection_token():
+    row = SimpleNamespace(token_sealed=secrets.seal("ghp_abc"))
+    assert connection_token(row) == "ghp_abc"
+
+
+def test_a_row_sealed_under_another_key_reads_as_none():
+    row = SimpleNamespace(token_sealed=secrets.seal("ghp_abc"))
+    with patch.object(secrets.Config, "SECRET_KEY", "a-different-instance"):
+        assert connection_token(row) is None
+
+
+def test_a_left_over_plain_text_token_is_not_used_as_one():
+    """A value that never went through seal is refused, not passed to the
+    forge as if it were the token: a reader sees None, never the raw bytes."""
+    assert connection_token(SimpleNamespace(token_sealed="plain-text-token")) is None

tests/test_integration_backup_file_ordering.py

@@ -0,0 +1,94 @@
+"""Real-Postgres check that the full export is ORDERED, so the backup file
+deduplicates between runs (milestone 523).
+
+Only Postgres shows this. An unordered select returns rows in heap order, and
+an UPDATE writes the row's new version at the end of the heap — so before
+export_full_backup ordered its queries, editing one old note moved it to the
+bottom of `notes`. Every byte after its old place shifted, and a restic run
+over the next backup stored most of the table again. A mocked session returns
+whatever order it was handed and could never catch it.
+"""
+import io
+import tarfile
+
+import pytest
+import pytest_asyncio
+from sqlalchemy import select
+
+from scribe.models import async_session
+from scribe.models.note import Note
+from scribe.models.user import User
+from scribe.services import backup
+from scribe.services.backup_file import write_archive_to
+from tests.helpers import ensure_user
+
+pytestmark = [pytest.mark.integration, pytest.mark.usefixtures("_dispose_engine")]
+
+USERNAME = "backup_file_ordering_owner"
+
+
+async def _purge() -> None:
+    async with async_session() as s:
+        for user in (await s.execute(
+            select(User).where(User.username == USERNAME)
+        )).scalars().all():
+            for note in (await s.execute(
+                select(Note).where(Note.user_id == user.id)
+            )).scalars().all():
+                await s.delete(note)
+            await s.delete(user)
+        await s.commit()
+
+
+@pytest_asyncio.fixture(autouse=True)
+async def _no_leftovers():
+    """Setup only — see the rule_version roundtrip module for why a teardown
+    here would break the next test's event loop."""
+    await _purge()
+
+
+@pytest_asyncio.fixture
+async def notes():
+    async with async_session() as s:
+        user = await ensure_user(s, USERNAME)
+        rows = [Note(user_id=user.id, title=f"ordering {n}", body="b" * 100)
+                for n in range(5)]
+        s.add_all(rows)
+        await s.commit()
+        ids = [n.id for n in rows]
+    yield ids
+    await _purge()
+
+
+def _sections(payload: dict) -> dict[str, bytes]:
+    buf = io.BytesIO()
+    write_archive_to(payload, buf, app_version="test")
+    buf.seek(0)
+    with tarfile.open(fileobj=buf, mode="r:") as tar:
+        return {m.name: tar.extractfile(m).read() for m in tar.getmembers()
+                if m.name != "manifest.json"}
+
+
+async def _edit_first(ids: list[int]) -> None:
+    async with async_session() as s:
+        note = await s.get(Note, ids[0])
+        note.body = "edited after the first export"
+        await s.commit()
+
+
+async def test_an_edited_row_keeps_its_place(notes):
+    await _edit_first(notes)
+    payload = await backup.export_full_backup()
+    exported = [row["id"] for row in payload["notes"]]
+    assert exported == sorted(exported)
+    mine = [i for i in exported if i in set(notes)]
+    assert mine == notes
+
+
+async def test_two_exports_of_unchanged_data_are_the_same_bytes(notes):
+    await _edit_first(notes)
+    first = _sections(await backup.export_full_backup())
+    second = _sections(await backup.export_full_backup())
+    assert first.keys() == second.keys()
+    for name in first:
+        assert first[name] == second[name], name

tests/test_integration_backup_file_restore.py

@@ -0,0 +1,103 @@
+"""Real-Postgres proof that the scheduled backup's FILE restores (milestone
+523): a payload written as the tar, read back through `scribe.restore`'s
+loader, and fed to the real `restore_full_backup`.
+
+The unit lane already shows the tar hands back the dict it was given. What
+only this lane shows is that the dict it hands back is one the restore
+accepts — every id remapped to the restored rows, nothing dropped — and that
+the command-line restore refuses a database that already has accounts.
+"""
+import pytest
+import pytest_asyncio
+from sqlalchemy import select
+
+from scribe.models import async_session
+from scribe.models.milestone import Milestone
+from scribe.models.note import Note
+from scribe.models.project import Project
+from scribe.models.user import User
+from scribe.restore import load_backup, run
+from scribe.services import backup
+from scribe.services.backup_file import write_archive
+from tests.helpers import ensure_user
+
+pytestmark = [pytest.mark.integration, pytest.mark.usefixtures("_dispose_engine")]
+
+OWNER = "backup_file_restore_owner"
+RESTORED = "backup_file_restore_restored"
+
+
+async def _purge(username: str) -> None:
+    async with async_session() as s:
+        for user in (await s.execute(select(User).where(User.username == username))).scalars().all():
+            for model in (Note, Milestone, Project):
+                for row in (await s.execute(
+                    select(model).where(model.user_id == user.id)
+                )).scalars().all():
+                    await s.delete(row)
+            await s.flush()
+            await s.delete(user)
+        await s.commit()
+
+
+@pytest_asyncio.fixture(autouse=True)
+async def _no_leftovers():
+    """Setup only — see the rule_version roundtrip module for why."""
+    await _purge(RESTORED)
+    await _purge(OWNER)
+
+
+@pytest_asyncio.fixture
+async def written(tmp_path):
+    """A project with a milestone and a task in it, exported with the real
+    row builders, renamed so the restore mints a separate user, and written
+    as the backup tar."""
+    async with async_session() as s:
+        user = await ensure_user(s, OWNER)
+        project = Project(user_id=user.id, title="Restore drill")
+        s.add(project)
+        await s.flush()
+        milestone = Milestone(user_id=user.id, project_id=project.id, title="Plan")
+        s.add(milestone)
+        await s.flush()
+        note = Note(user_id=user.id, project_id=project.id, milestone_id=milestone.id,
+                    title="Prove the restore", body="From the tar, not the JSON.",
+                    status="todo", note_type="note", task_kind="work")
+        s.add(note)
+        await s.commit()
+        payload = {
+            "version": backup.BACKUP_VERSION,
+            "scope": "full",
+            "users": backup._user_rows([user]),
+            "projects": backup._project_rows([project]),
+            "milestones": backup._milestone_rows([milestone]),
+            "notes": backup._note_rows([note]),
+        }
+    payload["users"][0]["username"] = RESTORED
+    from datetime import datetime, timezone
+    path = write_archive(payload, tmp_path, datetime.now(timezone.utc), app_version="test")
+    yield path
+    await _purge(RESTORED)
+    await _purge(OWNER)
+
+
+async def test_the_tar_restores_with_every_id_remapped(written):
+    await backup.restore_full_backup(load_backup(written))
+    async with async_session() as s:
+        user = (await s.execute(select(User).where(User.username == RESTORED))).scalars().one()
+        project = (await s.execute(select(Project).where(Project.user_id == user.id))).scalars().one()
+        milestone = (await s.execute(select(Milestone).where(Milestone.user_id == user.id))).scalars().one()
+        note = (await s.execute(select(Note).where(Note.user_id == user.id))).scalars().one()
+    assert project.title == "Restore drill"
+    assert (note.project_id, note.milestone_id) == (project.id, milestone.id)
+    assert note.body == "From the tar, not the JSON."
+    assert note.status == "todo"
+
+
+async def test_the_command_refuses_a_database_with_accounts(written):
+    """The source user exists, so this database is not empty: the command
+    must stop before restoring anything."""
+    assert await run(written) == 3
+    async with async_session() as s:
+        restored = (await s.execute(select(User).where(User.username == RESTORED))).scalars().all()
+    assert restored == []

tests/test_integration_forge_keyring.py

@@ -169,3 +169,29 @@ async def test_env_config_serves_admins_only_and_rows_beat_it(seeded):
         assert len(selector.adapters) == 1
         assert selector.adapters[0]._token == "row-tok"
         await delete_connection(admin, row.id)
+
+
+@pytest.mark.integration
+async def test_the_token_is_stored_sealed_and_a_key_change_drops_the_row(seeded):
+    """What a copy of the database holds is ciphertext; the reads still get
+    the token; and a row this instance can no longer open serves nothing
+    rather than sending ciphertext to the forge as a token."""
+    from sqlalchemy import text
+
+    from scribe.services import secrets
+
+    owner = seeded["owner"]
+    row = await create_connection(owner, kind="gitea", base_url=GITEA, token="plain-tok")
+    await update_connection(owner, row.id, token="rotated-tok")
+    async with async_session() as s:
+        stored = (await s.execute(
+            text("SELECT token_sealed FROM forge_connections WHERE id = :id"), {"id": row.id}
+        )).scalar_one()
+    assert "rotated-tok" not in stored
+    assert secrets.unseal(stored) == "rotated-tok"
+
+    selector = await get_forges(owner, seeded["project"])
+    assert [a._token for a in selector.adapters] == ["rotated-tok"]
+
+    with patch.object(secrets.Config, "SECRET_KEY", "a-different-instance"):
+        assert not (await get_forges(owner, seeded["project"])).configured

tests/test_services_backup_destination.py

@@ -0,0 +1,179 @@
+"""The backup destination (services/backup_destination.py) and the sealing it
+keeps the S3 secret under (services/secrets.py).
+
+No database: the admin settings are patched, the folder is a tmp_path, and the
+bucket is an httpx.MockTransport. What these pin is the contract a person
+meets in Settings — a config that is incomplete says what is missing, a place
+that can't take a backup says why, and the secret never leaves sealed.
+"""
+import asyncio
+import io
+from datetime import datetime, timezone
+from types import SimpleNamespace
+from unittest.mock import patch
+
+import httpx
+import pytest
+
+from scribe.services import backup, backup_destination, secrets
+from scribe.services.backup_destination import (
+    BackupConfig, DestinationError, check_destination, deliver,
+)
+from scribe.services.backup_file import read_archive
+
+AT = datetime(2026, 10, 15, 2, 0, 0, tzinfo=timezone.utc)
+PAYLOAD = {
+    "version": 25, "scope": "full", "exported_at": AT.isoformat(),
+    "users": [{"id": 1, "username": "ada"}], "notes": [],
+}
+S3 = BackupConfig(
+    destination="s3", s3_endpoint="http://archives:8081", s3_bucket="scribe",
+    s3_access_key="AKIA", s3_region="us-east-1",
+)
+
+
+# ── Sealing ──────────────────────────────────────────────────────────────────
+
+def test_a_sealed_secret_opens_and_is_not_the_plain_text():
+    token = secrets.seal("wJalrXUtnFEMI/K7MDENG")
+    assert "wJalrXUtnFEMI" not in token
+    assert secrets.unseal(token) == "wJalrXUtnFEMI/K7MDENG"
+
+
+def test_another_secret_key_cannot_open_it():
+    token = secrets.seal("s3cret")
+    with patch.object(secrets.Config, "SECRET_KEY", "a-different-instance"):
+        assert secrets.unseal(token) is None
+
+
+def test_unseal_of_nothing_or_garbage_is_none():
+    assert secrets.unseal("") is None
+    assert secrets.unseal("not a fernet token") is None
+
+
+def test_the_secret_setting_never_reaches_a_backup():
+    rows = [SimpleNamespace(user_id=1, key=k, value="v") for k in (
+        "backup_config", backup_destination.SECRET_SETTING, "theme",
+    )]
+    keys = [r["key"] for r in backup._setting_rows(rows)]
+    assert keys == ["backup_config", "theme"]
+
+
+def test_load_secret_tells_a_missing_secret_from_one_it_cannot_open():
+    async def stored(value):
+        async def get(key, default=""):
+            return value
+        with patch.object(backup_destination, "get_admin_setting", get):
+            return await backup_destination.load_secret()
+
+    assert asyncio.run(stored("")) == (None, False)
+    assert asyncio.run(stored(secrets.seal("k"))) == ("k", True)
+    assert asyncio.run(stored("sealed-elsewhere")) == (None, True)
+
+
+# ── Config ───────────────────────────────────────────────────────────────────
+
+def test_config_round_trips_and_defaults_to_off():
+    assert BackupConfig().enabled is False
+    assert BackupConfig.from_dict(S3.to_dict()) == S3
+
+
+def test_a_damaged_config_still_loads():
+    config = BackupConfig.from_dict({
+        "hour": "25", "destination": "ftp", "enabled": "yes", "unknown": 1,
+        "s3_bucket": "  scribe  ",
+    })
+    assert config.hour == backup_destination.DEFAULT_HOUR
+    assert config.destination == "folder"
+    assert config.enabled is False
+    assert config.s3_bucket == "scribe"
+
+
+def test_an_incomplete_s3_config_names_what_is_missing():
+    said = BackupConfig(destination="s3").problems(has_secret=False)
+    assert said == [
+        "The S3 address must start with http:// or https://.",
+        "The bucket name is missing.",
+        "The access key is missing.",
+        "The secret key is missing.",
+    ]
+    assert S3.problems(has_secret=True) == []
+
+
+def test_a_relative_folder_is_refused():
+    assert BackupConfig(folder_path="backups").problems(has_secret=False)
+
+
+# ── Folder ───────────────────────────────────────────────────────────────────
+
+def test_deliver_to_a_folder_adds_one_dated_file(tmp_path):
+    config = BackupConfig(folder_path=str(tmp_path))
+    got = asyncio.run(deliver(PAYLOAD, config, None, AT, app_version="v"))
+    assert got.name == "scribe-20261015T020000Z.tar"
+    assert [p.name for p in tmp_path.iterdir()] == [got.name]
+    assert got.size_bytes == (tmp_path / got.name).stat().st_size
+    with open(tmp_path / got.name, "rb") as f:
+        assert read_archive(f) == PAYLOAD
+
+
+def test_a_folder_that_isnt_mounted_says_so(tmp_path):
+    config = BackupConfig(folder_path=str(tmp_path / "not-mounted"))
+    with pytest.raises(DestinationError, match="isn't there"):
+        asyncio.run(deliver(PAYLOAD, config, None, AT, app_version="v"))
+
+
+def test_checking_a_folder_leaves_nothing_behind(tmp_path):
+    (tmp_path / "scribe-20261014T020000Z.tar").write_bytes(b"x")
+    config = BackupConfig(folder_path=str(tmp_path))
+    said = asyncio.run(check_destination(config, None))
+    assert "holds 1 Scribe backup" in said
+    assert [p.name for p in tmp_path.iterdir()] == ["scribe-20261014T020000Z.tar"]
+
+
+# ── S3 ───────────────────────────────────────────────────────────────────────
+
+def test_deliver_to_s3_puts_the_dated_file_and_keeps_no_copy():
+    seen = {}
+
+    def handler(request: httpx.Request):
+        seen["method"], seen["url"] = request.method, str(request.url)
+        seen["body"] = request.read()
+        return httpx.Response(200)
+
+    got = asyncio.run(deliver(PAYLOAD, S3, "secret", AT, app_version="v",
+                              transport=httpx.MockTransport(handler)))
+    assert seen["method"] == "PUT"
+    assert seen["url"] == "http://archives:8081/scribe/scribe-20261015T020000Z.tar"
+    assert read_archive(io.BytesIO(seen["body"])) == PAYLOAD
+    assert got.size_bytes == len(seen["body"])
+
+
+def test_a_refused_upload_is_a_destination_error_in_words():
+    body = b"<Error><Code>AccessDenied</Code><Message>no</Message></Error>"
+    transport = httpx.MockTransport(lambda r: httpx.Response(403, content=body))
+    with pytest.raises(DestinationError, match="the key isn't allowed to do that"):
+        asyncio.run(deliver(PAYLOAD, S3, "secret", AT, app_version="v", transport=transport))
+
+
+def test_s3_without_its_secret_sends_nothing():
+    def handler(request):
+        raise AssertionError("nothing should be sent")
+
+    with pytest.raises(DestinationError, match="secret key is missing"):
+        asyncio.run(deliver(PAYLOAD, S3, None, AT, app_version="v",
+                            transport=httpx.MockTransport(handler)))
+
+
+def test_checking_s3_only_lists():
+    xml = (b'<ListBucketResult xmlns="http://s3.amazonaws.com/doc/2006-03-01/">'
+           b"<Contents><Key>scribe-20261014T020000Z.tar</Key><Size>9</Size>"
+           b"<LastModified>x</LastModified></Contents></ListBucketResult>")
+    methods = []
+
+    def handler(request):
+        methods.append(request.method)
+        return httpx.Response(200, content=xml)
+
+    said = asyncio.run(check_destination(S3, "secret", transport=httpx.MockTransport(handler)))
+    assert methods == ["GET"]
+    assert "holds 1 Scribe backup" in said

tests/test_services_backup_file.py

@@ -0,0 +1,201 @@
+"""Unit tests for the backup file (services/backup_file.py).
+
+The file exists to be deduplicated between runs by a tool like restic, so
+most of what is tested here is a property of BYTES: what stays identical
+when nothing changed, and how little moves when one row does. A round trip
+alone would pass for a gzip of one big JSON document, which is exactly the
+shape this file replaced.
+"""
+import io
+import tarfile
+from datetime import datetime, timezone
+from unittest.mock import patch
+
+import pytest
+
+from scribe.services import backup_file
+from scribe.services.backup_file import (
+    ArchiveError, archive_name, is_archive, read_archive, write_archive,
+    write_archive_to,
+)
+
+AT = datetime(2026, 10, 15, 3, 0, 0, tzinfo=timezone.utc)
+
+
+def _payload(**over) -> dict:
+    payload = {
+        "version": 25,
+        "scope": "full",
+        "exported_at": "2026-10-15T03:00:00+00:00",
+        "_security_notice": "hashed passwords inside",
+        "_not_included": ["api_keys", "note_embeddings"],
+        "users": [{"id": 1, "username": "ada", "role": "admin"}],
+        "notes": [
+            {"id": n, "user_id": 1, "title": f"note {n}", "body": "x" * 200,
+             "tags": ["a", "b"]}
+            for n in range(1, 51)
+        ],
+        "settings": [],
+    }
+    payload.update(over)
+    return payload
+
+
+def _members(payload: dict) -> dict[str, bytes]:
+    buf = io.BytesIO()
+    write_archive_to(payload, buf, app_version="2026.10.15.0300")
+    buf.seek(0)
+    with tarfile.open(fileobj=buf, mode="r:") as tar:
+        return {m.name: tar.extractfile(m).read() for m in tar.getmembers()}
+
+
+def test_the_name_is_dated_in_utc_and_sorts():
+    assert archive_name(AT) == "scribe-20261015T030000Z.tar"
+    local = datetime(2026, 10, 14, 23, 0, 0).astimezone()
+    assert archive_name(local).endswith("Z.tar")
+
+
+def test_round_trip_returns_the_payload_restore_takes():
+    payload = _payload()
+    buf = io.BytesIO()
+    write_archive_to(payload, buf, app_version="v")
+    buf.seek(0)
+    assert read_archive(buf) == payload
+
+
+def test_notices_stay_in_the_manifest_not_as_sections():
+    """`_not_included` is a list, but of table names; it must not become a
+    member that a restore would read as rows."""
+    names = set(_members(_payload()))
+    assert names == {"manifest.json", "notes.jsonl", "settings.jsonl", "users.jsonl"}
+
+
+def test_members_come_in_one_order_with_fixed_headers():
+    buf = io.BytesIO()
+    write_archive_to(_payload(), buf, app_version="v")
+    buf.seek(0)
+    with tarfile.open(fileobj=buf, mode="r:") as tar:
+        members = tar.getmembers()
+    assert [m.name for m in members] == [
+        "manifest.json", "notes.jsonl", "settings.jsonl", "users.jsonl",
+    ]
+    for m in members:
+        assert (m.mtime, m.uid, m.gid, m.uname, m.gname, m.mode) == (
+            0, 0, 0, "", "", 0o644,
+        )
+
+
+def test_an_unchanged_payload_gives_identical_sections_on_another_run():
+    """The run's own time and the app version live in the manifest and
+    nowhere else, so a second run over unchanged data repeats every section
+    byte for byte."""
+    first = _members(_payload())
+    second = _members(_payload(exported_at="2026-10-16T03:00:00+00:00"))
+    assert first["manifest.json"] != second["manifest.json"]
+    for name in first:
+        if name != "manifest.json":
+            assert first[name] == second[name], name
+
+
+def test_key_order_in_a_row_does_not_change_its_bytes():
+    rows = _payload()["notes"]
+    shuffled = [dict(reversed(list(r.items()))) for r in rows]
+    assert _members(_payload(notes=shuffled))["notes.jsonl"] == \
+        _members(_payload())["notes.jsonl"]
+
+
+def test_editing_one_row_changes_one_line_of_one_member():
+    before = _members(_payload())
+    notes = _payload()["notes"]
+    notes[24] = {**notes[24], "body": "edited"}
+    after = _members(_payload(notes=notes))
+
+    changed = [n for n in before if before[n] != after[n]]
+    assert changed == ["notes.jsonl"]
+    old_lines = before["notes.jsonl"].splitlines()
+    new_lines = after["notes.jsonl"].splitlines()
+    assert len(old_lines) == len(new_lines)
+    assert [i for i, (a, b) in enumerate(zip(old_lines, new_lines)) if a != b] == [24]
+
+
+def test_a_new_row_appends_to_the_end_of_its_member():
+    """Rows come in primary-key order, so a table that grows grows at its
+    tail: everything already sent is a prefix of the new member."""
+    before = _members(_payload())["notes.jsonl"]
+    notes = _payload()["notes"] + [{"id": 51, "user_id": 1, "title": "new"}]
+    after = _members(_payload(notes=notes))["notes.jsonl"]
+    assert after.startswith(before)
+    assert after.count(b"\n") == before.count(b"\n") + 1
+
+
+def test_the_file_is_an_uncompressed_tar():
+    buf = io.BytesIO()
+    write_archive_to(_payload(), buf, app_version="v")
+    raw = buf.getvalue()
+    assert raw[257:262] == b"ustar"
+    assert b'"title":"note 7"' in raw
+
+
+def test_is_archive_tells_a_tar_from_the_json_download():
+    buf = io.BytesIO()
+    write_archive_to(_payload(), buf, app_version="v")
+    buf.seek(0)
+    assert is_archive(buf)
+    assert buf.tell() == 0
+    assert not is_archive(io.BytesIO(b'{"version": 25, "users": []}'))
+
+
+def test_write_archive_lands_under_the_dated_name_only(tmp_path):
+    path = write_archive(_payload(), tmp_path, AT, app_version="v")
+    assert path == tmp_path / "scribe-20261015T030000Z.tar"
+    assert [p.name for p in tmp_path.iterdir()] == [path.name]
+    with open(path, "rb") as f:
+        assert read_archive(f) == _payload()
+
+
+def test_a_failed_write_leaves_nothing_behind(tmp_path):
+    """Neither a half backup under the final name (a watcher would count it
+    as arrived) nor a stray part file."""
+    with patch.object(backup_file, "write_archive_to", side_effect=OSError("disk full")):
+        with pytest.raises(OSError):
+            write_archive(_payload(), tmp_path, AT, app_version="v")
+    assert list(tmp_path.iterdir()) == []
+
+
+def _rewrite(members: dict[str, bytes]) -> io.BytesIO:
+    buf = io.BytesIO()
+    with tarfile.open(fileobj=buf, mode="w", format=tarfile.USTAR_FORMAT) as tar:
+        for name, data in members.items():
+            info = tarfile.TarInfo(name)
+            info.size = len(data)
+            tar.addfile(info, io.BytesIO(data))
+    buf.seek(0)
+    return buf
+
+
+def test_a_missing_section_is_an_error_not_a_thinner_restore():
+    members = _members(_payload())
+    del members["notes.jsonl"]
+    with pytest.raises(ArchiveError, match="notes.jsonl"):
+        read_archive(_rewrite(members))
+
+
+def test_a_section_cut_short_is_an_error():
+    members = _members(_payload())
+    members["notes.jsonl"] = b"\n".join(members["notes.jsonl"].splitlines()[:10]) + b"\n"
+    with pytest.raises(ArchiveError, match="10 rows"):
+        read_archive(_rewrite(members))
+
+
+def test_a_newer_format_is_refused_by_name():
+    members = _members(_payload())
+    members["manifest.json"] = members["manifest.json"].replace(
+        b'"format_version": 1', b'"format_version": 99',
+    )
+    with pytest.raises(ArchiveError, match="upgrade"):
+        read_archive(_rewrite(members))
+
+
+def test_garbage_is_not_an_archive():
+    with pytest.raises(ArchiveError):
+        read_archive(io.BytesIO(b"not a tar at all" * 100))

tests/test_services_backup_job.py

@@ -0,0 +1,223 @@
+"""The scheduled backup job (services/backup_job.py) and the admin surface
+over it. No database: admin settings are an in-memory dict, and the export
+and delivery are stand-ins, so what is under test is the job's own contract —
+it records every outcome, keeps a bounded history, runs only at its hour,
+and never runs twice at once.
+"""
+import asyncio
+from datetime import datetime, timedelta, timezone
+from unittest.mock import patch
+
+import pytest
+
+from scribe.services import backup_destination, backup_job
+from scribe.services.backup_destination import (
+    BackupConfig, Delivered, DestinationError, apply_changes,
+)
+
+NOW = datetime(2026, 10, 15, 2, 0, 0, tzinfo=timezone.utc)
+
+
+class _Settings(dict):
+    async def get(self, key, default=""):
+        return self.get_value(key, default)
+
+    def get_value(self, key, default):
+        return dict.get(self, key, default)
+
+    async def set(self, key, value):
+        self[key] = value
+
+
+def _patched(settings, deliver=None, config=BackupConfig(enabled=True)):
+    async def load_config():
+        return config
+
+    async def load_secret():
+        return (None, False)
+
+    async def export():
+        return {"version": 25, "users": []}
+
+    async def delivered(payload, config, secret, at, **kw):
+        return Delivered("scribe-20261015T020000Z.tar", 1234, "/backups")
+
+    return [
+        patch.object(backup_job, "get_admin_setting", settings.get),
+        patch.object(backup_job, "set_admin_setting", settings.set),
+        patch.object(backup_job, "load_config", load_config),
+        patch.object(backup_job, "load_secret", load_secret),
+        patch.object(backup_job, "deliver", deliver or delivered),
+        patch("scribe.services.backup.export_full_backup", export),
+    ]
+
+
+def _run(coro, patches):
+    for p in patches:
+        p.start()
+    try:
+        return asyncio.run(coro)
+    finally:
+        for p in reversed(patches):
+            p.stop()
+
+
+def test_a_good_run_is_kept_with_its_file_and_size():
+    settings = _Settings()
+    run = _run(backup_job.run_backup("manual"), _patched(settings))
+    assert run["status"] == "ok"
+    assert (run["file"], run["size_bytes"], run["trigger"]) == (
+        "scribe-20261015T020000Z.tar", 1234, "manual",
+    )
+    runs = _run(backup_job.list_runs(), _patched(settings))
+    assert runs == [run]
+
+
+def test_a_failed_run_is_kept_with_its_reason_in_words():
+    async def refuse(*a, **kw):
+        raise DestinationError("/backups isn't there.")
+
+    settings = _Settings()
+    run = _run(backup_job.run_backup("scheduled"), _patched(settings, deliver=refuse))
+    assert run["status"] == "failed"
+    assert run["error"] == "/backups isn't there."
+
+
+def test_an_unexpected_error_is_a_failed_run_not_a_crash():
+    async def boom(*a, **kw):
+        raise RuntimeError("disk on fire")
+
+    settings = _Settings()
+    run = _run(backup_job.run_backup("scheduled"), _patched(settings, deliver=boom))
+    assert run["status"] == "failed"
+    assert "disk on fire" in run["error"]
+
+
+def test_a_run_that_hangs_is_abandoned_at_the_deadline():
+    async def hang(*a, **kw):
+        await asyncio.sleep(10)
+
+    settings = _Settings()
+    with patch.object(backup_job, "RUN_DEADLINE", timedelta(seconds=0.05)):
+        run = _run(backup_job.run_backup("scheduled"), _patched(settings, deliver=hang))
+    assert run["status"] == "failed"
+    assert "didn't finish" in run["error"]
+    assert not backup_job.is_running()
+
+
+def test_the_history_keeps_only_the_newest_runs():
+    settings = _Settings()
+
+    async def many():
+        for _ in range(backup_job.KEEP_RUNS + 5):
+            await backup_job.run_backup("scheduled")
+        return await backup_job.list_runs()
+
+    runs = _run(many(), _patched(settings))
+    assert len(runs) == backup_job.KEEP_RUNS
+    assert runs[0]["started_at"] >= runs[-1]["started_at"]
+
+
+def test_a_second_run_while_one_is_going_is_refused():
+    started = asyncio.Event()
+    release = asyncio.Event()
+
+    async def slow(*a, **kw):
+        started.set()
+        await release.wait()
+        return Delivered("x.tar", 1, "/backups")
+
+    async def both():
+        first = asyncio.create_task(backup_job.run_backup("manual"))
+        await started.wait()
+        with pytest.raises(backup_job.BackupBusy):
+            await backup_job.run_backup("manual")
+        release.set()
+        return await first
+
+    run = _run(both(), _patched(_Settings(), deliver=slow))
+    assert run["status"] == "ok"
+
+
+@pytest.mark.parametrize("enabled,hour,ran", [
+    (True, NOW.hour, True),
+    (True, (NOW.hour + 1) % 24, False),
+    (False, NOW.hour, False),
+])
+def test_the_hourly_tick_runs_only_when_on_and_at_its_hour(enabled, hour, ran):
+    calls = []
+
+    async def fake_run(trigger):
+        calls.append(trigger)
+
+    class _Now(datetime):
+        @classmethod
+        def now(cls, tz=None):
+            return NOW
+
+    async def config():
+        return BackupConfig(enabled=enabled, hour=hour)
+
+    with patch.object(backup_job, "load_config", config), \
+         patch.object(backup_job, "run_backup", fake_run), \
+         patch.object(backup_job, "datetime", _Now):
+        asyncio.run(backup_job._hourly_check())
+    assert calls == (["scheduled"] if ran else [])
+
+
+# ── The page's verdict ───────────────────────────────────────────────────────
+
+def _ok(at):
+    return {"status": "ok", "finished_at": at.isoformat()}
+
+
+def test_a_recent_success_is_not_overdue():
+    s = backup_job.summarise([_ok(NOW - timedelta(hours=3))], enabled=True, now=NOW)
+    assert s["overdue"] is False
+    assert s["last_success"]["status"] == "ok"
+
+
+def test_failures_after_an_old_success_are_overdue():
+    runs = [{"status": "failed", "finished_at": NOW.isoformat()},
+            _ok(NOW - timedelta(hours=30))]
+    s = backup_job.summarise(runs, enabled=True, now=NOW)
+    assert s["overdue"] is True
+    assert s["last_run"]["status"] == "failed"
+
+
+def test_off_is_never_overdue_and_nothing_run_yet_is_not_late():
+    assert backup_job.summarise([], enabled=False, now=NOW)["overdue"] is False
+    assert backup_job.summarise([], enabled=True, now=NOW)["overdue"] is False
+    failed_only = [{"status": "failed", "finished_at": NOW.isoformat()}]
+    assert backup_job.summarise(failed_only, enabled=True, now=NOW)["overdue"] is True
+
+
+# ── What an admin sends ──────────────────────────────────────────────────────
+
+def test_changes_apply_on_top_of_the_current_config():
+    config = apply_changes(BackupConfig(), {"enabled": True, "hour": "5", "s3_secret": "x"})
+    assert (config.enabled, config.hour, config.destination) == (True, 5, "folder")
+
+
+@pytest.mark.parametrize("data,said", [
+    ({"hour": 24}, "from 0 to 23"),
+    ({"hour": "two"}, "whole number"),
+    ({"destination": "ftp"}, "folder or an S3 bucket"),
+    ({"enabled": "yes"}, "true or false"),
+])
+def test_a_typed_mistake_is_refused_with_the_reason(data, said):
+    with pytest.raises(ValueError, match=said):
+        apply_changes(BackupConfig(), data)
+
+
+def test_routes_and_scheduler_are_wired():
+    from scribe.app import create_app
+    from scribe.routes import admin as admin_routes
+
+    for name in ("get_backups", "update_backups", "check_backups_destination", "run_backup_now"):
+        assert callable(getattr(admin_routes, name))
+    rules = {r.rule for r in create_app().url_map.iter_rules()}
+    assert {"/api/admin/backups", "/api/admin/backups/check", "/api/admin/backups/run"} <= rules
+    assert callable(backup_job.start_backup_scheduler)
+    assert callable(backup_job.stop_backup_scheduler)
+    assert backup_destination.SECRET_SETTING == "backup_s3_secret"

tests/test_services_backup_s3.py

@@ -0,0 +1,112 @@
+"""The backup destination's S3 client (services/backup_s3.py): signatures match AWS's
+published Signature Version 4 examples, and the calls read S3's answers."""
+import asyncio
+from datetime import datetime, timezone
+
+import httpx
+import pytest
+
+from scribe.services import backup_s3 as s3
+
+# AWS's example credentials and moment ("Signature Calculations for the
+# Authorization Header", Amazon S3 API reference).
+KEY_ID = "AKIAIOSFODNN7EXAMPLE"
+SECRET = "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"
+AT = datetime(2013, 5, 24, tzinfo=timezone.utc)
+
+
+def _signature(headers):
+    return headers["Authorization"].rsplit("Signature=", 1)[1]
+
+
+def test_aws_example_get_object():
+    h = s3.sign("GET", "https://examplebucket.s3.amazonaws.com/test.txt", {"Range": "bytes=0-9"},
+                s3.EMPTY_SHA256, KEY_ID, SECRET, "us-east-1", AT)
+    assert "SignedHeaders=host;range;x-amz-content-sha256;x-amz-date," in h["Authorization"]
+    assert "Credential=AKIAIOSFODNN7EXAMPLE/20130524/us-east-1/s3/aws4_request" in h["Authorization"]
+    assert _signature(h) == "f0e8bdb87c964420e857bd35b5d6ed310bd44f0170aba48dd91039c6036bdb41"
+
+
+def test_aws_example_list_objects():
+    h = s3.sign("GET", "https://examplebucket.s3.amazonaws.com/?max-keys=2&prefix=J", {},
+                s3.EMPTY_SHA256, KEY_ID, SECRET, "us-east-1", AT)
+    assert _signature(h) == "34b48302e7b5fa45bde8084f4b7868a86f0a534bc59db6670ed5711ef69dc6f7"
+
+
+def test_a_changed_secret_changes_the_signature():
+    a = s3.sign("GET", "https://examplebucket.s3.amazonaws.com/test.txt", {},
+                s3.EMPTY_SHA256, KEY_ID, SECRET, "us-east-1", AT)
+    b = s3.sign("GET", "https://examplebucket.s3.amazonaws.com/test.txt", {},
+                s3.EMPTY_SHA256, KEY_ID, SECRET + "x", "us-east-1", AT)
+    assert _signature(a) != _signature(b)
+
+
+TARGET = s3.S3Target("http://archives:8081/s3", "scribe", KEY_ID, SECRET)
+
+
+def test_paths_are_path_style():
+    assert TARGET.url("scribe-20261012T000000Z.tar") == "http://archives:8081/s3/scribe/scribe-20261012T000000Z.tar"
+    assert TARGET.url() == "http://archives:8081/s3/scribe/"
+
+
+def test_put_sends_the_file_signed(tmp_path):
+    f = tmp_path / "b.tar"
+    f.write_bytes(b"x" * 5000)
+    seen = {}
+
+    def handler(request: httpx.Request):
+        seen["method"], seen["url"] = request.method, str(request.url)
+        seen["body"] = request.read()
+        seen["auth"] = request.headers["authorization"]
+        seen["sha"] = request.headers["x-amz-content-sha256"]
+        return httpx.Response(200)
+
+    asyncio.run(s3.put_file(TARGET, "b.tar", f, transport=httpx.MockTransport(handler)))
+    assert seen["method"] == "PUT"
+    assert seen["url"] == "http://archives:8081/s3/scribe/b.tar"
+    assert seen["body"] == b"x" * 5000
+    assert seen["sha"] == s3.file_sha256(f)
+    assert seen["auth"].startswith("AWS4-HMAC-SHA256 Credential=AKIAIOSFODNN7EXAMPLE/")
+
+
+def test_a_refusal_says_why(tmp_path):
+    f = tmp_path / "b.tar"
+    f.write_bytes(b"x")
+    body = b"<Error><Code>SignatureDoesNotMatch</Code><Message>no</Message></Error>"
+    transport = httpx.MockTransport(lambda r: httpx.Response(403, content=body))
+    with pytest.raises(s3.S3Error, match="HTTP 403 SignatureDoesNotMatch: the secret key is wrong"):
+        asyncio.run(s3.put_file(TARGET, "b.tar", f, transport=transport))
+
+
+def test_unreachable_says_so(tmp_path):
+    f = tmp_path / "b.tar"
+    f.write_bytes(b"x")
+
+    def handler(request):
+        raise httpx.ConnectError("refused", request=request)
+
+    with pytest.raises(s3.S3Error, match="Couldn't reach http://archives:8081/s3"):
+        asyncio.run(s3.put_file(TARGET, "b.tar", f, transport=httpx.MockTransport(handler)))
+
+
+def test_list_reads_contents():
+    xml = b"""<?xml version="1.0" encoding="UTF-8"?>
+<ListBucketResult xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
+  <Name>scribe</Name><KeyCount>2</KeyCount>
+  <Contents><Key>scribe-20261011T000000Z.tar</Key><Size>1200</Size>
+    <LastModified>2026-10-11T00:00:01.000Z</LastModified></Contents>
+  <Contents><Key>scribe-20261012T000000Z.tar</Key><Size>1300</Size>
+    <LastModified>2026-10-12T00:00:01.000Z</LastModified></Contents>
+</ListBucketResult>"""
+    seen = {}
+
+    def handler(request):
+        seen["params"] = dict(request.url.params)
+        return httpx.Response(200, content=xml)
+
+    got = asyncio.run(s3.list_objects(TARGET, "scribe-", transport=httpx.MockTransport(handler)))
+    assert seen["params"] == {"list-type": "2", "max-keys": "1000", "prefix": "scribe-"}
+    assert got == [
+        s3.StoredObject("scribe-20261011T000000Z.tar", 1200, "2026-10-11T00:00:01.000Z"),
+        s3.StoredObject("scribe-20261012T000000Z.tar", 1300, "2026-10-12T00:00:01.000Z"),
+    ]

uv.lock

@@ -1508,6 +1508,7 @@ dependencies = [
     { name = "asyncpg" },
     { name = "bcrypt" },
     { name = "caldav" },
+    { name = "cryptography" },
     { name = "fastembed" },
     { name = "httpx" },
     { name = "hypercorn" },
@@ -1533,6 +1534,7 @@ requires-dist = [
     { name = "asyncpg", specifier = ">=0.29" },
     { name = "bcrypt", specifier = ">=4.0" },
     { name = "caldav", specifier = ">=1.3" },
+    { name = "cryptography" },
     { name = "fastembed", specifier = ">=0.4" },
     { name = "httpx", specifier = ">=0.27" },
     { name = "hypercorn", specifier = ">=0.17" },