Merge pull request 'DB maintenance + health observability + Postgres integration CI lane' (#72) from dev into main

First integration run proved the lane works (run_maintenance test passed against
real Postgres), but the health test failed with 'Future attached to a different
loop': pytest-asyncio uses a fresh loop per test while the app's module-level
engine pools a connection from the prior test's loop. Dispose the engine in each
test's teardown so the next test starts with an empty pool on its own loop.

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

.forgejo/workflows/ci.yml

@@ -142,7 +142,75 @@ jobs:
           uv pip install --python /opt/venv/bin/python -e ".[dev]"
 
       - name: Run tests
-        run: /opt/venv/bin/python -m pytest tests/ -q
+        # Integration tests (real Postgres) run in the `integration` job below.
+        run: /opt/venv/bin/python -m pytest tests/ -q -m "not integration"
+
+  # Real-Postgres lane (family rule 6). Exercises the async SQLAlchemy connection
+  # path the unit stubs can't reach — the un-awaited execution_options regression
+  # that made every VACUUM report 0/6 lived here. Like `test`, it runs for
+  # visibility and does NOT gate the build.
+  #
+  # Job key stays separator-free ("integration"): act_runner derives the service-
+  # container name from the (truncated) job display name and the discovery step
+  # filters `docker ps` by it. Service hostnames aren't routable on this runner,
+  # so the step resolves the Postgres container's bridge IP. No `name:` on purpose.
+  integration:
+    if: github.ref == 'refs/heads/dev' || github.ref == 'refs/heads/main' || startsWith(github.ref, 'refs/tags/v')
+    runs-on: python-ci
+    container:
+      image: git.fabledsword.com/bvandeusen/ci-python:3.14
+    env:
+      # Config + the module engine read these at import time. DATABASE_URL itself
+      # is built from the discovered service IP in the run step.
+      SECRET_KEY: ci_integration_placeholder
+    services:
+      postgres:
+        image: postgres:16-alpine
+        env:
+          POSTGRES_USER: scribe
+          POSTGRES_PASSWORD: ci_integration
+          POSTGRES_DB: scribe_test
+        options: >-
+          --health-cmd "pg_isready -U scribe"
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 10
+    steps:
+      - uses: actions/checkout@v6
+      - name: Create virtual environment
+        run: uv venv /opt/venv
+      - name: Install package with dev deps
+        run: |
+          uv pip install --python /opt/venv/bin/python setuptools wheel
+          uv pip install --python /opt/venv/bin/python --no-build-isolation http-ece
+          uv pip install --python /opt/venv/bin/python -e ".[dev]"
+      - name: Integration suite (resolve service IP, migrate, test)
+        run: |
+          set -eux
+          echo "=== container landscape (diagnostic for the name filter) ==="
+          docker ps -a --format '{{.ID}} {{.Image}} -> {{.Names}}'
+          PG=$(docker ps --filter "name=integration" --filter "ancestor=postgres:16-alpine" -q | head -n1)
+          test -n "$PG"
+          PG_IP=$(docker inspect -f '{{range .NetworkSettings.Networks}}{{.IPAddress}}{{end}}' "$PG")
+          test -n "$PG_IP"
+          export DATABASE_URL="postgresql+asyncpg://scribe:ci_integration@${PG_IP}:5432/scribe_test"
+          # Wait for Postgres to accept connections (busybox sh — the runner
+          # default — has no bash /dev/tcp, so use Python).
+          /opt/venv/bin/python - "$PG_IP" <<'PY'
+          import socket, sys, time
+          for _ in range(30):
+              try:
+                  socket.create_connection((sys.argv[1], 5432), timeout=2).close()
+                  break
+              except OSError:
+                  time.sleep(1)
+          else:
+              sys.exit("postgres did not become reachable")
+          PY
+          # Real migrations build the schema; the maintenance tests then run
+          # VACUUM (ANALYZE) and read pg_stat_user_tables against it.
+          /opt/venv/bin/alembic upgrade head
+          /opt/venv/bin/python -m pytest tests/ -v -m integration
 
   build:
     name: Build & push image

frontend/src/views/SettingsView.vue

@@ -150,6 +150,33 @@ const serverMarketplaceUrl = ref('');
 const adminMarketplaceUrl = ref('');
 const savingMarketplaceUrl = ref(false);
 const marketplaceUrlSaved = ref(false);
+
+// DB maintenance (admin) — daily targeted VACUUM (ANALYZE).
+interface DbMaintTableResult { table: string; ok: boolean; elapsed_ms: number; error: string | null }
+interface DbMaintRun { started_at: string; elapsed_ms: number; tables: DbMaintTableResult[] }
+const dbMaintEnabled = ref(true);
+const dbMaintHour = ref(4);
+const dbMaintLastRun = ref<DbMaintRun | null>(null);
+const savingDbMaint = ref(false);
+const dbMaintSaved = ref(false);
+const runningDbMaint = ref(false);
+interface DbTableHealth {
+  table: string; live: number; dead: number; dead_pct: number;
+  total_bytes: number; mod_since_analyze: number;
+  last_vacuum: string | null; last_analyze: string | null;
+}
+interface DbHealth { db_bytes: number; tables: DbTableHealth[] }
+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
@@ -448,6 +475,19 @@ onMounted(async () => {
     // not configured yet
   }
 
+  // DB maintenance config (admin only — endpoint is admin-gated).
+  if (authStore.isAdmin) {
+    try {
+      const dm = await apiGet<{ enabled: boolean; hour: number; last_run: DbMaintRun | null }>("/api/admin/db-maintenance");
+      dbMaintEnabled.value = dm.enabled;
+      dbMaintHour.value = dm.hour;
+      dbMaintLastRun.value = dm.last_run;
+    } catch {
+      // leave defaults
+    }
+    await loadDbHealth();
+  }
+
   // Load admin settings
   if (authStore.isAdmin) {
     try {
@@ -697,6 +737,54 @@ async function saveMarketplaceUrl() {
   }
 }
 
+async function saveDbMaintenance() {
+  savingDbMaint.value = true;
+  dbMaintSaved.value = false;
+  try {
+    await apiPut("/api/admin/db-maintenance", {
+      enabled: dbMaintEnabled.value,
+      hour: dbMaintHour.value,
+    });
+    dbMaintSaved.value = true;
+    setTimeout(() => (dbMaintSaved.value = false), 2000);
+  } catch (e) {
+    const body = (e as { body?: { error?: string } }).body;
+    toastStore.show(body?.error || "Failed to save maintenance settings", "error");
+  } finally {
+    savingDbMaint.value = false;
+  }
+}
+
+async function loadDbHealth() {
+  loadingHealth.value = true;
+  try {
+    dbHealth.value = await apiGet<DbHealth>("/api/admin/db-maintenance/health");
+  } catch {
+    // leave previous value
+  } finally {
+    loadingHealth.value = false;
+  }
+}
+
+async function runDbMaintenanceNow() {
+  runningDbMaint.value = true;
+  try {
+    const summary = await apiPost<DbMaintRun>("/api/admin/db-maintenance/run", {});
+    dbMaintLastRun.value = summary;
+    const failed = summary.tables.filter((t) => !t.ok).length;
+    toastStore.show(
+      failed ? `Maintenance ran with ${failed} error(s)` : "Maintenance complete",
+      failed ? "error" : "success",
+    );
+    await loadDbHealth(); // reflect the dead-tuple drop
+  } catch (e) {
+    const body = (e as { body?: { error?: string } }).body;
+    toastStore.show(body?.error || "Maintenance run failed", "error");
+  } finally {
+    runningDbMaint.value = false;
+  }
+}
+
 async function handleRestoreFile(event: Event) {
   const file = (event.target as HTMLInputElement).files?.[0];
   if (!file) return;
@@ -1724,6 +1812,86 @@ function formatUserDate(iso: string): string {
         </div>
       </section>
 
+      <section class="settings-section full-width">
+        <h2>Database maintenance</h2>
+        <p class="section-desc">
+          A daily <code>VACUUM (ANALYZE)</code> over the high-churn tables (logs, notifications,
+          tokens, notes, version history) — on top of Postgres autovacuum — to reclaim space left
+          by the nightly cleanup sweeps and keep query plans fresh. Runs at the hour below (UTC),
+          just after trash purge.
+        </p>
+        <div class="checkbox-field">
+          <label>
+            <input type="checkbox" v-model="dbMaintEnabled" />
+            Run scheduled maintenance daily
+          </label>
+        </div>
+        <div class="field url-field">
+          <label for="db-maint-hour">Run hour (UTC)</label>
+          <select id="db-maint-hour" v-model.number="dbMaintHour" class="input">
+            <option v-for="h in 24" :key="h - 1" :value="h - 1">
+              {{ String(h - 1).padStart(2, '0') }}:00
+            </option>
+          </select>
+        </div>
+        <div class="actions">
+          <button class="btn-save" @click="saveDbMaintenance" :disabled="savingDbMaint">
+            {{ savingDbMaint ? "Saving..." : "Save" }}
+          </button>
+          <button class="btn-save btn-secondary" @click="runDbMaintenanceNow" :disabled="runningDbMaint">
+            {{ runningDbMaint ? "Running..." : "Run now" }}
+          </button>
+          <span v-if="dbMaintSaved" class="saved-msg">Saved!</span>
+        </div>
+        <div v-if="dbMaintLastRun" class="db-maint-last">
+          <span class="db-maint-last-label">
+            Last run {{ new Date(dbMaintLastRun.started_at).toLocaleString() }}
+            · {{ dbMaintLastRun.elapsed_ms }}ms
+          </span>
+          <ul class="db-maint-table-list">
+            <li v-for="t in dbMaintLastRun.tables" :key="t.table" :class="{ 'dm-failed': !t.ok }">
+              <code>{{ t.table }}</code>
+              <span class="dm-status">{{ t.ok ? `✓ ${t.elapsed_ms}ms` : `✗ ${t.error}` }}</span>
+            </li>
+          </ul>
+        </div>
+
+        <div class="db-health">
+          <h3 class="subsection-label">
+            Table health
+            <span v-if="dbHealth" class="db-health-total">· database {{ formatBytes(dbHealth.db_bytes) }}</span>
+          </h3>
+          <p class="field-hint">
+            Dead-tuple ratio is bloat — rows left by updates/deletes not yet reclaimed.
+            Above {{ DEAD_PCT_WARN }}% on a large table means autovacuum is falling behind;
+            consider adding it to the maintenance set.
+          </p>
+          <p v-if="loadingHealth && !dbHealth" class="field-hint">Loading…</p>
+          <div v-else-if="dbHealth" class="db-health-scroll">
+            <table class="db-health-table">
+              <thead>
+                <tr>
+                  <th>Table</th><th class="num">Size</th><th class="num">Live</th>
+                  <th class="num">Dead</th><th class="num">Dead %</th>
+                  <th>Last vacuum</th><th>Last analyze</th>
+                </tr>
+              </thead>
+              <tbody>
+                <tr v-for="t in dbHealth.tables" :key="t.table" :class="{ 'dh-warn': t.dead_pct >= DEAD_PCT_WARN }">
+                  <td><code>{{ t.table }}</code></td>
+                  <td class="num">{{ formatBytes(t.total_bytes) }}</td>
+                  <td class="num">{{ t.live.toLocaleString() }}</td>
+                  <td class="num">{{ t.dead.toLocaleString() }}</td>
+                  <td class="num">{{ t.dead_pct }}%</td>
+                  <td>{{ t.last_vacuum ? new Date(t.last_vacuum).toLocaleString() : "—" }}</td>
+                  <td>{{ t.last_analyze ? new Date(t.last_analyze).toLocaleString() : "—" }}</td>
+                </tr>
+              </tbody>
+            </table>
+          </div>
+        </div>
+      </section>
+
       <section class="settings-section full-width">
         <h2>Email / SMTP</h2>
         <p class="section-desc">Configure SMTP to enable email notifications for all users.</p>
@@ -2352,6 +2520,58 @@ function formatUserDate(iso: string): string {
 }
 .btn-secondary:hover:not(:disabled) { background: var(--color-action-secondary-hover); }
 .btn-secondary:disabled { opacity: 0.6; cursor: default; }
+
+/* DB maintenance last-run summary */
+.db-maint-last { margin-top: 1rem; }
+.db-maint-last-label {
+  display: block;
+  font-size: 0.8rem;
+  color: var(--color-text-muted);
+  margin-bottom: 0.4rem;
+}
+.db-maint-table-list {
+  list-style: none;
+  margin: 0;
+  padding: 0;
+  display: flex;
+  flex-direction: column;
+  gap: 0.2rem;
+}
+.db-maint-table-list li {
+  display: flex;
+  justify-content: space-between;
+  gap: 1rem;
+  font-size: 0.82rem;
+  padding: 0.2rem 0;
+}
+.db-maint-table-list .dm-status { color: var(--color-text-muted); }
+.db-maint-table-list li.dm-failed .dm-status { color: var(--color-danger); }
+
+/* DB table-health readout */
+.db-health { margin-top: 1.5rem; }
+.db-health-total { color: var(--color-text-muted); font-weight: 400; }
+.db-health-scroll { overflow-x: auto; margin-top: 0.5rem; }
+.db-health-table {
+  width: 100%;
+  border-collapse: collapse;
+  font-size: 0.82rem;
+}
+.db-health-table th, .db-health-table td {
+  text-align: left;
+  padding: 0.35rem 0.6rem;
+  border-bottom: 1px solid var(--color-border);
+  white-space: nowrap;
+}
+.db-health-table th {
+  font-size: 0.72rem;
+  text-transform: uppercase;
+  letter-spacing: 0.04em;
+  color: var(--color-text-muted);
+  font-weight: 500;
+}
+.db-health-table td.num, .db-health-table th.num { text-align: right; font-variant-numeric: tabular-nums; }
+.db-health-table tr.dh-warn td { color: var(--color-warning); }
+.db-health-table tr.dh-warn td:first-child code { color: var(--color-warning); }
 .btn-warn:hover:not(:disabled) {
   background: var(--color-warning);
   color: #fff;

pyproject.toml

@@ -36,6 +36,9 @@ where = ["src"]
 [tool.pytest.ini_options]
 asyncio_mode = "auto"
 testpaths = ["tests"]
+markers = [
+    "integration: requires a real Postgres database (runs only in the CI integration lane)",
+]
 
 [tool.ruff]
 line-length = 120

src/scribe/app.py

@@ -187,6 +187,15 @@ def create_app() -> Quart:
         from scribe.services.trash_scheduler import start_trash_scheduler
         start_trash_scheduler(asyncio.get_running_loop())
 
+        # DB maintenance scheduler (daily targeted VACUUM ANALYZE, default 04:00 UTC)
+        from scribe.services.db_maintenance_scheduler import (
+            get_maintenance_hour,
+            start_db_maintenance_scheduler,
+        )
+        start_db_maintenance_scheduler(
+            asyncio.get_running_loop(), await get_maintenance_hour()
+        )
+
         # 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.
@@ -203,6 +212,10 @@ def create_app() -> Quart:
         stop_version_pinning_scheduler()
         from scribe.services.trash_scheduler import stop_trash_scheduler
         stop_trash_scheduler()
+        from scribe.services.db_maintenance_scheduler import (
+            stop_db_maintenance_scheduler,
+        )
+        stop_db_maintenance_scheduler()
         from scribe.services.diagnostics import stop_diagnostics
         stop_diagnostics()
 

src/scribe/routes/admin.py

@@ -21,7 +21,12 @@ from scribe.services.backup import (
 from scribe.services.email import SMTP_SETTING_KEYS, get_base_url, get_smtp_config, is_smtp_configured, send_test_email
 from scribe.services.logging import get_logs, get_log_stats, log_audit
 from scribe.services.notifications import send_invitation_email
-from scribe.services.settings import set_setting, set_settings_batch
+from scribe.services.settings import (
+    get_admin_setting,
+    set_admin_setting,
+    set_setting,
+    set_settings_batch,
+)
 
 admin_bp = Blueprint("admin", __name__, url_prefix="/api/admin")
 
@@ -204,6 +209,69 @@ async def update_base_url():
     return jsonify({"status": "ok"})
 
 
+@admin_bp.route("/db-maintenance", methods=["GET"])
+@admin_required
+async def get_db_maintenance():
+    """Current DB-maintenance config + the last run's summary."""
+    from scribe.services.db_maintenance import get_last_run
+    from scribe.services.db_maintenance_scheduler import (
+        get_maintenance_hour,
+        is_maintenance_enabled,
+    )
+    return jsonify({
+        "enabled": await is_maintenance_enabled(),
+        "hour": await get_maintenance_hour(),
+        "last_run": await get_last_run(),
+    })
+
+
+@admin_bp.route("/db-maintenance/health", methods=["GET"])
+@admin_required
+async def get_db_maintenance_health():
+    """Read-only per-table bloat/health stats from Postgres + total DB size."""
+    from scribe.services.db_maintenance import get_table_health
+    return jsonify(await get_table_health())
+
+
+@admin_bp.route("/db-maintenance", methods=["PUT"])
+@admin_required
+async def update_db_maintenance():
+    """Set whether scheduled maintenance runs and at what UTC hour."""
+    from scribe.services.db_maintenance_scheduler import reschedule_db_maintenance
+    data = await request.get_json() or {}
+    enabled = bool(data.get("enabled", True))
+    try:
+        hour = int(data.get("hour", 4))
+    except (TypeError, ValueError):
+        return jsonify({"error": "hour must be an integer 0–23"}), 400
+    if not 0 <= hour <= 23:
+        return jsonify({"error": "hour must be between 0 and 23"}), 400
+
+    await set_admin_setting("db_maintenance_enabled", "true" if enabled else "false")
+    await set_admin_setting("db_maintenance_hour", str(hour))
+    reschedule_db_maintenance(hour)
+    uid = get_current_user_id()
+    await log_audit(
+        "db_maintenance_config", user_id=uid, username=g.user.username,
+        ip_address=request.remote_addr, details={"enabled": enabled, "hour": hour},
+    )
+    return jsonify({"status": "ok", "enabled": enabled, "hour": hour})
+
+
+@admin_bp.route("/db-maintenance/run", methods=["POST"])
+@admin_required
+async def run_db_maintenance_now():
+    """Run a VACUUM (ANALYZE) sweep immediately and return its summary."""
+    from scribe.services.db_maintenance import run_maintenance
+    uid = get_current_user_id()
+    await log_audit(
+        "db_maintenance_run", user_id=uid, username=g.user.username,
+        ip_address=request.remote_addr,
+    )
+    summary = await run_maintenance()
+    return jsonify(summary)
+
+
 @admin_bp.route("/invitations", methods=["POST"])
 @admin_required
 async def create_invite():

src/scribe/services/db_maintenance.py

@@ -0,0 +1,175 @@
+"""Basic Postgres maintenance — targeted VACUUM (ANALYZE).
+
+Postgres autovacuum already reclaims dead tuples and refreshes planner stats
+in the background. This adds a daily, off-hours top-up over the handful of
+tables the retention/purge sweeps churn hardest (log retention, notification
+purge, auth-token purge, trash purge, version pruning) — so the bloat those
+bulk DELETEs leave behind is collected promptly and the planner keeps good
+stats. It is deliberately narrow; the rest of the schema is left to autovacuum.
+
+Security: table names cannot be parameterized in SQL, so VACUUM is only ever
+issued against names from the hardcoded MAINTENANCE_TABLES allowlist — there is
+no path from user input to the table name.
+"""
+from __future__ import annotations
+
+import json
+import logging
+import time
+from datetime import datetime, timezone
+
+from sqlalchemy import text
+
+from scribe.models import async_session, engine
+from scribe.services.settings import get_admin_setting, set_admin_setting
+
+logger = logging.getLogger(__name__)
+
+# High-churn tables — exactly the ones the scheduled delete sweeps hit:
+#   app_logs              ← log retention (hourly)
+#   notifications         ← read-notification purge (hourly)
+#   password_reset_tokens ← auth-token purge (daily)
+#   invitation_tokens     ← auth-token purge (daily)
+#   notes                 ← trash purge (daily) + general churn
+#   note_versions         ← version-pinning prune (daily)
+MAINTENANCE_TABLES: tuple[str, ...] = (
+    "app_logs",
+    "notifications",
+    "password_reset_tokens",
+    "invitation_tokens",
+    "notes",
+    "note_versions",
+)
+
+_LAST_RUN_KEY = "db_maintenance_last_run"
+
+
+async def run_maintenance(tables: tuple[str, ...] | list[str] | None = None) -> dict:
+    """Run VACUUM (ANALYZE) over the allowlisted high-churn tables.
+
+    VACUUM cannot run inside a transaction block, so this uses a dedicated
+    AUTOCOMMIT connection rather than the usual session. Each table is
+    vacuumed independently; one failure is logged and does not abort the rest.
+
+    Returns a summary dict (also persisted as the db_maintenance_last_run
+    admin setting) of shape:
+        {"started_at": iso, "elapsed_ms": int, "tables": [
+            {"table": str, "ok": bool, "elapsed_ms": int, "error": str|None}
+        ]}
+    """
+    # Only ever operate on the closed allowlist — never an arbitrary name.
+    requested = tuple(tables) if tables is not None else MAINTENANCE_TABLES
+    targets = [t for t in requested if t in MAINTENANCE_TABLES]
+
+    started = time.monotonic()
+    started_at = datetime.now(timezone.utc).isoformat()
+    results: list[dict] = []
+
+    async with engine.connect() as conn:
+        # On AsyncConnection, execution_options() is a coroutine and MUST be
+        # awaited (it returns the connection with AUTOCOMMIT set — VACUUM can't
+        # run inside a transaction block).
+        autocommit = await conn.execution_options(isolation_level="AUTOCOMMIT")
+        for table in targets:
+            t0 = time.monotonic()
+            try:
+                # table is from the allowlist above — safe to interpolate.
+                await autocommit.exec_driver_sql(f"VACUUM (ANALYZE) {table}")
+                results.append({
+                    "table": table,
+                    "ok": True,
+                    "elapsed_ms": int((time.monotonic() - t0) * 1000),
+                    "error": None,
+                })
+            except Exception as exc:  # noqa: BLE001 — record per-table, keep going
+                logger.exception("VACUUM (ANALYZE) failed for %s", table)
+                results.append({
+                    "table": table,
+                    "ok": False,
+                    "elapsed_ms": int((time.monotonic() - t0) * 1000),
+                    "error": str(exc),
+                })
+
+    summary = {
+        "started_at": started_at,
+        "elapsed_ms": int((time.monotonic() - started) * 1000),
+        "tables": results,
+    }
+    ok = sum(1 for r in results if r["ok"])
+    logger.info(
+        "DB maintenance: vacuumed %d/%d table(s) in %dms",
+        ok, len(results), summary["elapsed_ms"],
+    )
+    try:
+        await set_admin_setting(_LAST_RUN_KEY, json.dumps(summary))
+    except Exception:  # noqa: BLE001 — persisting the summary is best-effort
+        logger.warning("Failed to persist db_maintenance last-run summary", exc_info=True)
+    return summary
+
+
+# Read-only table-health query against Postgres' own statistics views. The key
+# signal is the dead-tuple ratio (dead / (live + dead)) — that IS bloat; a high
+# ratio on a large table means autovacuum isn't keeping up. Timestamps confirm
+# maintenance is actually running on each table.
+_HEALTH_SQL = text("""
+    SELECT
+      relname AS table_name,
+      n_live_tup AS live,
+      n_dead_tup AS dead,
+      CASE WHEN n_live_tup + n_dead_tup > 0
+           THEN round(100.0 * n_dead_tup / (n_live_tup + n_dead_tup), 1)
+           ELSE 0 END AS dead_pct,
+      pg_total_relation_size(relid) AS total_bytes,
+      n_mod_since_analyze AS mod_since_analyze,
+      GREATEST(last_vacuum, last_autovacuum) AS last_vacuum,
+      GREATEST(last_analyze, last_autoanalyze) AS last_analyze
+    FROM pg_stat_user_tables
+    ORDER BY n_dead_tup DESC, total_bytes DESC
+""")
+
+
+def _iso(value) -> str | None:
+    return value.isoformat() if value is not None else None
+
+
+async def get_table_health() -> dict:
+    """Per-table health from Postgres statistics + the total database size.
+
+    Read-only (system catalogs only). Returns:
+        {"db_bytes": int, "tables": [
+            {"table": str, "live": int, "dead": int, "dead_pct": float,
+             "total_bytes": int, "mod_since_analyze": int,
+             "last_vacuum": iso|None, "last_analyze": iso|None}
+        ]}
+    """
+    async with async_session() as session:
+        db_bytes = (
+            await session.execute(text("SELECT pg_database_size(current_database())"))
+        ).scalar() or 0
+        rows = (await session.execute(_HEALTH_SQL)).mappings().all()
+
+    tables = [
+        {
+            "table": r["table_name"],
+            "live": int(r["live"] or 0),
+            "dead": int(r["dead"] or 0),
+            "dead_pct": float(r["dead_pct"] or 0),
+            "total_bytes": int(r["total_bytes"] or 0),
+            "mod_since_analyze": int(r["mod_since_analyze"] or 0),
+            "last_vacuum": _iso(r["last_vacuum"]),
+            "last_analyze": _iso(r["last_analyze"]),
+        }
+        for r in rows
+    ]
+    return {"db_bytes": int(db_bytes), "tables": tables}
+
+
+async def get_last_run() -> dict | None:
+    """Return the most recent run summary, or None if it has never run."""
+    raw = await get_admin_setting(_LAST_RUN_KEY, "")
+    if not raw:
+        return None
+    try:
+        return json.loads(raw)
+    except (ValueError, TypeError):
+        return None

src/scribe/services/db_maintenance_scheduler.py

@@ -0,0 +1,107 @@
+"""Daily APScheduler cron for basic DB maintenance (targeted VACUUM ANALYZE).
+
+Mirrors trash_scheduler.py: a single global BackgroundScheduler job bridges
+into the asyncio loop to run the async maintenance. Scheduled for 04:00 UTC by
+default — after the 03:30 trash purge — so it collects the dead tuples that
+night's delete sweeps leave behind.
+
+Two things are operator-tunable from the admin Settings card:
+  - db_maintenance_enabled ("true"/"false") — checked at fire time, so toggling
+    it needs no reschedule.
+  - db_maintenance_hour ("0".."23", UTC) — the cron hour. Changing it calls
+    reschedule_db_maintenance() so the live job moves without a restart.
+"""
+from __future__ import annotations
+
+import asyncio
+import logging
+
+from apscheduler.schedulers.background import BackgroundScheduler
+from apscheduler.triggers.cron import CronTrigger
+
+from scribe.services.settings import get_admin_setting
+
+logger = logging.getLogger(__name__)
+
+_JOB_ID = "db_maintenance_vacuum"
+_DEFAULT_HOUR = 4
+
+_scheduler: BackgroundScheduler | None = None
+_loop: asyncio.AbstractEventLoop | None = None
+
+
+async def get_maintenance_hour() -> int:
+    """The configured run-hour (UTC, 0–23), clamped; default 04:00."""
+    raw = await get_admin_setting("db_maintenance_hour", str(_DEFAULT_HOUR))
+    try:
+        hour = int(raw)
+    except (TypeError, ValueError):
+        return _DEFAULT_HOUR
+    return hour if 0 <= hour <= 23 else _DEFAULT_HOUR
+
+
+async def is_maintenance_enabled() -> bool:
+    """Whether the scheduled run is enabled (default on)."""
+    return (await get_admin_setting("db_maintenance_enabled", "true")) != "false"
+
+
+def _run_maintenance_threadsafe() -> None:
+    """APScheduler invokes this from a worker thread; bridge into the loop."""
+    if _loop is None:
+        logger.warning("db maintenance scheduler: no loop registered")
+        return
+
+    async def _runner():
+        try:
+            if not await is_maintenance_enabled():
+                logger.debug("db maintenance: disabled, skipping scheduled run")
+                return
+            from scribe.services.db_maintenance import run_maintenance
+            await run_maintenance()
+        except Exception:
+            logger.exception("db maintenance run failed")
+
+    asyncio.run_coroutine_threadsafe(_runner(), _loop)
+
+
+def start_db_maintenance_scheduler(
+    loop: asyncio.AbstractEventLoop, hour: int = _DEFAULT_HOUR
+) -> None:
+    """Start the daily job. `hour` is the configured UTC run-hour, resolved by
+    the caller (which has an async context) via get_maintenance_hour() — passed
+    in rather than read here so we never block the event loop at startup. The
+    job's enabled-gate is re-checked at every fire, so only the hour is needed
+    up front."""
+    global _scheduler, _loop
+    if _scheduler is not None:
+        return
+    _loop = loop
+    hour = hour if 0 <= hour <= 23 else _DEFAULT_HOUR
+    _scheduler = BackgroundScheduler()
+    _scheduler.add_job(
+        _run_maintenance_threadsafe,
+        trigger=CronTrigger(hour=hour, minute=0, timezone="UTC"),
+        id=_JOB_ID,
+        replace_existing=True,
+    )
+    _scheduler.start()
+    logger.info("DB maintenance scheduler started (daily %02d:00 UTC)", hour)
+
+
+def reschedule_db_maintenance(hour: int) -> None:
+    """Move the live job to a new UTC hour (called when the admin changes it)."""
+    if _scheduler is None:
+        return
+    hour = hour if 0 <= hour <= 23 else _DEFAULT_HOUR
+    _scheduler.reschedule_job(
+        _JOB_ID, trigger=CronTrigger(hour=hour, minute=0, timezone="UTC")
+    )
+    logger.info("DB maintenance scheduler rescheduled to %02d:00 UTC", hour)
+
+
+def stop_db_maintenance_scheduler() -> None:
+    global _scheduler
+    if _scheduler is not None:
+        _scheduler.shutdown(wait=False)
+        _scheduler = None
+        logger.info("DB maintenance scheduler stopped")

src/scribe/services/settings.py

@@ -26,6 +26,23 @@ async def get_admin_setting(key: str, default: str = "") -> str:
         return setting.value if setting and setting.value else default
 
 
+async def set_admin_setting(key: str, value: str) -> None:
+    """Write an instance-global setting onto the first admin account.
+
+    The write-side counterpart to get_admin_setting, for non-per-user settings
+    written outside a request context (e.g. a scheduler persisting its last-run
+    summary) where get_current_user_id() isn't available.
+    """
+    async with async_session() as session:
+        admin_id = (
+            await session.execute(
+                select(User.id).where(User.role == "admin").order_by(User.id)
+            )
+        ).scalars().first()
+    if admin_id is not None:
+        await set_setting(admin_id, key, value)
+
+
 async def get_setting(user_id: int, key: str, default: str = "") -> str:
     async with async_session() as session:
         result = await session.execute(

tests/conftest.py

@@ -12,8 +12,14 @@ import pytest
 
 
 @pytest.fixture(autouse=True)
-def _isolate_env(monkeypatch):
-    """Prevent tests from accidentally reading production env vars."""
+def _isolate_env(request, monkeypatch):
+    """Prevent unit tests from accidentally reading production env vars.
+
+    Integration tests (marked `integration`) are skipped here: they must use the
+    real DATABASE_URL injected by the CI integration lane, not the fake one.
+    """
+    if request.node.get_closest_marker("integration"):
+        return
     monkeypatch.setenv("DATABASE_URL", "postgresql+asyncpg://test:test@localhost/test")
     monkeypatch.setenv("SECRET_KEY", "test-secret-key")
     monkeypatch.setenv("OLLAMA_URL", "http://localhost:11434")

tests/test_integration_db_maintenance.py

@@ -0,0 +1,56 @@
+"""Real-Postgres integration tests for DB maintenance + health.
+
+These run only in the CI integration lane (a real Postgres service + schema
+built by `alembic upgrade head`). They exercise the actual async SQLAlchemy
+connection path that unit mocks cannot: the un-awaited
+`AsyncConnection.execution_options` regression (which made every VACUUM raise
+AttributeError, reporting 0/6) passes the unit suite but fails here.
+"""
+import pytest
+import pytest_asyncio
+
+from scribe.models import engine
+from scribe.services.db_maintenance import (
+    MAINTENANCE_TABLES,
+    get_table_health,
+    run_maintenance,
+)
+
+pytestmark = pytest.mark.integration
+
+
+@pytest_asyncio.fixture(autouse=True)
+async def _dispose_engine():
+    """Dispose the app's module-level engine after each test.
+
+    The engine pools asyncpg connections per event loop, but pytest-asyncio runs
+    each test on a fresh loop — so without this, test 2 gets handed test 1's
+    connection bound to a now-dead loop ("Future attached to a different loop").
+    Disposing in the test's own loop teardown clears the pool cleanly.
+    """
+    yield
+    await engine.dispose()
+
+
+@pytest.mark.asyncio
+async def test_run_maintenance_vacuums_real_tables():
+    summary = await run_maintenance()
+    assert summary["tables"], "no tables were vacuumed"
+    # Every allowlisted table exists after `alembic upgrade head`, so every
+    # VACUUM (ANALYZE) must succeed. With the un-awaited execution_options bug
+    # they would ALL fail with AttributeError — this is the guard.
+    failed = [t for t in summary["tables"] if not t["ok"]]
+    assert not failed, f"VACUUM failed for: {failed}"
+    assert {t["table"] for t in summary["tables"]} <= set(MAINTENANCE_TABLES)
+
+
+@pytest.mark.asyncio
+async def test_table_health_reports_real_stats():
+    health = await get_table_health()
+    assert health["db_bytes"] > 0
+    names = {t["table"] for t in health["tables"]}
+    # Core table built by migrations must show up in pg_stat_user_tables.
+    assert "notes" in names
+    for t in health["tables"]:
+        assert t["dead_pct"] >= 0
+        assert t["total_bytes"] >= 0

tests/test_routes_db_maintenance.py

@@ -0,0 +1,54 @@
+"""Structural tests for the DB-maintenance admin routes + scheduler surface."""
+
+
+def test_admin_handlers_callable():
+    from scribe.routes import admin as admin_routes
+    for name in (
+        "get_db_maintenance",
+        "update_db_maintenance",
+        "run_db_maintenance_now",
+        "get_db_maintenance_health",
+    ):
+        assert callable(getattr(admin_routes, name))
+
+
+def test_db_maintenance_routes_registered():
+    from scribe.app import create_app
+    app = create_app()
+    rules = {r.rule for r in app.url_map.iter_rules()}
+    assert "/api/admin/db-maintenance" in rules
+    assert "/api/admin/db-maintenance/run" in rules
+    assert "/api/admin/db-maintenance/health" in rules
+
+
+def test_scheduler_surface():
+    from scribe.services import db_maintenance_scheduler as sched
+    for fn in (
+        "start_db_maintenance_scheduler",
+        "stop_db_maintenance_scheduler",
+        "reschedule_db_maintenance",
+        "get_maintenance_hour",
+        "is_maintenance_enabled",
+    ):
+        assert callable(getattr(sched, fn)), f"scheduler missing {fn}"
+
+
+def test_service_surface_and_allowlist():
+    from scribe.services.db_maintenance import (
+        MAINTENANCE_TABLES,
+        get_last_run,
+        get_table_health,
+        run_maintenance,
+    )
+    assert callable(run_maintenance) and callable(get_last_run) and callable(get_table_health)
+    # The allowlist is a closed tuple covering exactly the high-churn tables.
+    assert isinstance(MAINTENANCE_TABLES, tuple)
+    assert set(MAINTENANCE_TABLES) == {
+        "app_logs", "notifications", "password_reset_tokens",
+        "invitation_tokens", "notes", "note_versions",
+    }
+
+
+def test_set_admin_setting_exists():
+    from scribe.services.settings import set_admin_setting
+    assert callable(set_admin_setting)

tests/test_services_db_maintenance.py

@@ -0,0 +1,124 @@
+"""Tests for services/db_maintenance.py — mocks the engine (no real DB).
+
+Verifies that VACUUM (ANALYZE) is issued once per allowlisted table, that the
+allowlist is closed (an arbitrary name can't be vacuumed), and that a single
+table failure doesn't abort the rest of the sweep.
+"""
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from scribe.services.db_maintenance import MAINTENANCE_TABLES, run_maintenance
+
+
+def _mock_engine(exec_side_effect=None):
+    """An engine whose connect() yields a conn with a recording exec_driver_sql."""
+    conn = MagicMock()
+    autocommit = MagicMock()
+    autocommit.exec_driver_sql = AsyncMock(side_effect=exec_side_effect)
+    # On AsyncConnection, execution_options() is a coroutine — mock it as async
+    # so the test exercises the real (awaited) call shape.
+    conn.execution_options = AsyncMock(return_value=autocommit)
+    cm = AsyncMock()
+    cm.__aenter__ = AsyncMock(return_value=conn)
+    cm.__aexit__ = AsyncMock(return_value=False)
+    engine = MagicMock()
+    engine.connect.return_value = cm
+    return engine, autocommit
+
+
+@pytest.mark.asyncio
+async def test_vacuums_each_allowlisted_table_once():
+    engine, autocommit = _mock_engine()
+    with patch("scribe.services.db_maintenance.engine", engine), \
+         patch("scribe.services.db_maintenance.set_admin_setting", AsyncMock()):
+        summary = await run_maintenance()
+
+    issued = [c.args[0] for c in autocommit.exec_driver_sql.await_args_list]
+    assert issued == [f"VACUUM (ANALYZE) {t}" for t in MAINTENANCE_TABLES]
+    assert all(r["ok"] for r in summary["tables"])
+    assert len(summary["tables"]) == len(MAINTENANCE_TABLES)
+    # AUTOCOMMIT was requested (and awaited) — VACUUM can't run in a transaction.
+    conn = engine.connect.return_value.__aenter__.return_value
+    conn.execution_options.assert_awaited_once_with(isolation_level="AUTOCOMMIT")
+
+
+@pytest.mark.asyncio
+async def test_allowlist_is_closed():
+    """A caller-supplied name not on the allowlist is silently ignored."""
+    engine, autocommit = _mock_engine()
+    with patch("scribe.services.db_maintenance.engine", engine), \
+         patch("scribe.services.db_maintenance.set_admin_setting", AsyncMock()):
+        summary = await run_maintenance(tables=["app_logs", "users; DROP TABLE notes"])
+
+    issued = [c.args[0] for c in autocommit.exec_driver_sql.await_args_list]
+    assert issued == ["VACUUM (ANALYZE) app_logs"]
+    assert [r["table"] for r in summary["tables"]] == ["app_logs"]
+
+
+@pytest.mark.asyncio
+async def test_one_table_failure_does_not_abort_the_rest():
+    # First table raises, the remaining tables still get vacuumed.
+    calls = {"n": 0}
+
+    def flaky(_sql):
+        calls["n"] += 1
+        if calls["n"] == 1:
+            raise RuntimeError("boom")
+        return MagicMock()
+
+    engine, autocommit = _mock_engine(exec_side_effect=flaky)
+    with patch("scribe.services.db_maintenance.engine", engine), \
+         patch("scribe.services.db_maintenance.set_admin_setting", AsyncMock()):
+        summary = await run_maintenance()
+
+    assert autocommit.exec_driver_sql.await_count == len(MAINTENANCE_TABLES)
+    assert summary["tables"][0]["ok"] is False
+    assert summary["tables"][0]["error"] == "boom"
+    assert all(r["ok"] for r in summary["tables"][1:])
+
+
+def _health_session(db_bytes, rows):
+    s = AsyncMock()
+    s.__aenter__ = AsyncMock(return_value=s)
+    s.__aexit__ = AsyncMock(return_value=False)
+    size_res = MagicMock()
+    size_res.scalar.return_value = db_bytes
+    rows_res = MagicMock()
+    rows_res.mappings.return_value.all.return_value = rows
+    s.execute = AsyncMock(side_effect=[size_res, rows_res])
+    return s
+
+
+@pytest.mark.asyncio
+async def test_table_health_shapes_rows_and_db_size():
+    from datetime import datetime, timezone
+
+    from scribe.services.db_maintenance import get_table_health
+    vac = datetime(2026, 6, 14, 4, 0, tzinfo=timezone.utc)
+    rows = [
+        {"table_name": "notes", "live": 1000, "dead": 300, "dead_pct": 23.1,
+         "total_bytes": 5_000_000, "mod_since_analyze": 50,
+         "last_vacuum": vac, "last_analyze": None},
+    ]
+    with patch("scribe.services.db_maintenance.async_session",
+               return_value=_health_session(42_000_000, rows)):
+        health = await get_table_health()
+
+    assert health["db_bytes"] == 42_000_000
+    t = health["tables"][0]
+    assert t["table"] == "notes"
+    assert t["dead_pct"] == 23.1
+    assert t["last_vacuum"] == vac.isoformat()
+    assert t["last_analyze"] is None  # null timestamp passes through as None
+
+
+@pytest.mark.asyncio
+async def test_summary_is_persisted_as_admin_setting():
+    engine, _ = _mock_engine()
+    setter = AsyncMock()
+    with patch("scribe.services.db_maintenance.engine", engine), \
+         patch("scribe.services.db_maintenance.set_admin_setting", setter):
+        await run_maintenance()
+    setter.assert_awaited_once()
+    assert setter.await_args.args[0] == "db_maintenance_last_run"