fix: require explicit SQLite migration preflight

docs/operations.md

@@ -302,14 +302,22 @@ their version, name, checksum, and application time in `schema_migrations`.
 `schema.sql` remains the current snapshot for fresh databases; migrations make
 existing databases converge on that snapshot and perform required backfills.
 Applied migration files are immutable: a checksum mismatch or a database newer
-than the installed package aborts startup instead of guessing. For an explicit
-operator run outside autoupdate (after independently ensuring the queue is
-quiet), use:
+than the installed package aborts startup instead of guessing. Ordinary queue,
+CLI, dashboard and webhook startup validates migration history without applying
+pending steps or the rolling schema snapshot. For an explicit operator run
+outside autoupdate, use:
 
 ```bash
 gab --db ~/.local/state/github-agent-bridge/bridge.sqlite3 migrate-db
 ```
 
+`migrate-db` refuses to run while any `pending`, `running` or
+`waiting_approval` job exists, even if the executor is paused. For an existing
+database it creates an online backup before applying the snapshot and versioned
+steps; use `--backup-dir` to choose the destination. A failed migration restores
+that backup before returning an error. Fresh database creation remains allowed
+without a backup because there is no prior state to preserve.
+
 Dashboard admins can run the same first-step workflow from the autoupdate notice:
 `Check now` refreshes and records the plan, `Apply update` runs the immediate
 safe subset, and `Complete reload` runs the recorded deferred reload once the

docs/persistence.md

@@ -25,9 +25,10 @@ domain repositories, the service owns one transaction and passes that
 connection to every participating repository; repositories must not open a
 nested independent transaction.
 
-Backup and restore are deliberately excluded. They use SQLite's backup API in
-`autoupdate.py` and remain specialized database operations rather than
-repository methods.
+Backup and restore are deliberately excluded from repositories. The specialized
+`backup_sqlite_database()` and `restore_sqlite_database()` operations use
+SQLite's backup API and are shared by autoupdate and the explicit migration
+command.
 
 Executor heartbeat writes, acknowledgement claims and streamed session-event
 writes treat `SQLITE_BUSY` and `SQLITE_LOCKED` as transient contention after the

src/github_agent_bridge/autoupdate.py

@@ -11,13 +11,20 @@
 
 from . import __version__
 from .models import utc_now
+from .persistence import (
+    ACTIVE_JOB_STATUSES,
+    Database,
+    StateRepository,
+    active_job_counts,
+    backup_sqlite_database,
+    restore_sqlite_database,
+)
 from .policy import Policy
 from .queue import JobQueue
 
 CommandRunner = Callable[[Sequence[str], Path | None], subprocess.CompletedProcess[str]]
 
 UPDATE_STATE_KEY = "autoupdate"
-ACTIVE_JOB_STATUSES = ("pending", "running", "waiting_approval")
 RISKY_PATH_PREFIXES = (
     "src/github_agent_bridge/cli.py",
     "src/github_agent_bridge/dispatch.py",
@@ -256,6 +263,34 @@ def update_queue_state(queue: JobQueue) -> dict[str, Any]:
     }
 
 
+def database_queue_state(
+    db: str | Path,
+    *,
+    migration_preflight: bool = False,
+) -> dict[str, Any]:
+    """Inspect queue state without triggering schema initialization or migration."""
+    database = Database(db)
+    if not database.path.exists():
+        return {
+            "active_counts": {status: 0 for status in ACTIVE_JOB_STATUSES},
+            "active_total": 0,
+            "executor_paused": False,
+        }
+    counts = active_job_counts(database)
+    executor_paused = StateRepository(database).executor_paused()
+    active_statuses = (
+        ACTIVE_JOB_STATUSES
+        if migration_preflight or not executor_paused
+        else ("running",)
+    )
+    active_counts = {status: int(counts.get(status, 0)) for status in active_statuses}
+    return {
+        "active_counts": active_counts,
+        "active_total": sum(active_counts.values()),
+        "executor_paused": executor_paused,
+    }
+
+
 def _service_plan_restarts_executor(service_plan: dict[str, Any]) -> bool:
     units = service_plan.get("units") if isinstance(service_plan.get("units"), dict) else {}
     executor_unit = str(units.get("executor") or DEFAULT_SYSTEMD_UNITS["executor"])
@@ -309,32 +344,12 @@ def _command_result(
 
 
 def _backup_sqlite_db(db: str | Path, backup_dir: str | Path | None = None) -> dict[str, Any]:
-    db_path = Path(db).expanduser()
     backup_root = Path(backup_dir).expanduser() if backup_dir else DEFAULT_BACKUP_DIR
-    backup_root.mkdir(parents=True, exist_ok=True)
-    timestamp = utc_now().replace(":", "").replace("-", "")
-    backup_path = backup_root / f"{db_path.stem}-{timestamp}.sqlite3"
-    with sqlite3.connect(db_path) as source, sqlite3.connect(backup_path) as target:
-        source.backup(target)
-    return {
-        "path": str(backup_path),
-        "created_at": utc_now(),
-        "source": str(db_path),
-        "size_bytes": backup_path.stat().st_size,
-    }
+    return backup_sqlite_database(db, backup_root)
 
 
 def _restore_sqlite_db(db: str | Path, backup_path: str | Path) -> dict[str, Any]:
-    db_path = Path(db).expanduser()
-    backup = Path(backup_path).expanduser()
-    with sqlite3.connect(backup) as source, sqlite3.connect(db_path) as target:
-        source.backup(target)
-    return {
-        "restored_at": utc_now(),
-        "source": str(backup),
-        "target": str(db_path),
-        "size_bytes": db_path.stat().st_size,
-    }
+    return restore_sqlite_database(db, backup_path)
 
 
 def default_migration_command(db: str | Path, *, python_bin: str = sys.executable) -> list[str]:
@@ -565,7 +580,7 @@ def apply_update_plan(
     migration_state: dict[str, Any] = {}
     service_plan = plan.get("service_plan") if isinstance(plan.get("service_plan"), dict) else {}
     queue_info = (
-        update_queue_state(JobQueue(db))
+        database_queue_state(db, migration_preflight=bool(migration_files))
         if db is not None
         else plan.get("queue") if isinstance(plan.get("queue"), dict) else {}
     )
@@ -612,7 +627,7 @@ def apply_update_plan(
             return result
 
     if migration_files and run_migrations:
-        queue_info = update_queue_state(JobQueue(db))
+        queue_info = database_queue_state(db, migration_preflight=True)
         result["queue"] = queue_info
         if int(queue_info.get("active_total") or 0):
             migration_state["status"] = "deferred"
@@ -796,12 +811,9 @@ def plan_update(
     systemd_units: dict[str, str] | None = None,
     runner: CommandRunner = _default_runner,
 ) -> dict[str, Any]:
-    queue = JobQueue(db)
     repo_path = Path(repo_dir).expanduser().resolve()
     current_tag = f"v{installed_version.lstrip('v')}"
     release = ReleaseInfo(tag_name=target_tag, source="explicit_target") if target_tag else latest_release(repo, gh_bin=gh_bin, runner=runner)
-    queue_state = update_queue_state(queue)
-    active_total = int(queue_state["active_total"])
 
     warnings: list[str] = []
     try:
@@ -812,6 +824,11 @@ def plan_update(
     classification = classify_changed_files(files)
     up_to_date = release.tag_name == current_tag
     migration_required = bool(classification["migration_files"])
+    queue_state = database_queue_state(
+        db,
+        migration_preflight=migration_required,
+    )
+    active_total = int(queue_state["active_total"])
     executor_reload_pending = False
     dashboard_restart_allowed = False
     executor_restart_allowed = False

src/github_agent_bridge/cli.py

@@ -7,13 +7,20 @@
 import os
 import re
 import shlex
+import sqlite3
 import subprocess
 import sys
 from pathlib import Path
 
 from . import feedback
 from .actors import backfill_trigger_actors
-from .autoupdate import apply_update_plan, complete_pending_reload, plan_update, record_update_plan
+from .autoupdate import (
+    DEFAULT_BACKUP_DIR,
+    apply_update_plan,
+    complete_pending_reload,
+    plan_update,
+    record_update_plan,
+)
 from .cancellation import cancel_running_job
 from .dashboard_data import inspect_db_read_only, list_jobs
 from .dispatch import FEEDBACK_LEARNING_RULES, GitHubClient, OpenClawDispatcher, RunMode, prompt_rule
@@ -23,9 +30,16 @@
 from .mcp import authenticate_token, create_token, list_tokens, revoke_token, serve_stdio
 from .observability import DEFAULT_PROCESS_SAMPLE_RETENTION_SECONDS, configure_sentry
 from .parser import decode_header_value, extract_body_text, is_github_notification_message, parse_auth_results
+from .persistence import (
+    Database,
+    active_job_counts,
+    backup_sqlite_database,
+    restore_sqlite_database,
+)
 from .policy import Policy, validate_policy_file
 from .queue import JobQueue
 from .reader import ImapConfig, ImapReader, imap_mailbox_arg
+from .sql.migrations import migration_history
 
 DEFAULT_DB = os.path.expanduser("~/.local/state/github-agent-bridge/bridge.sqlite3")
 DEFAULT_POLICY = os.path.expanduser("~/.config/github-agent-bridge/policy.json")
@@ -91,16 +105,41 @@ def cmd_init_db(args: argparse.Namespace) -> int:
 
 
 def cmd_migrate_db(args: argparse.Namespace) -> int:
-    queue = JobQueue(args.db)
+    db_path = Path(args.db).expanduser()
+    backup: dict | None = None
+    if db_path.exists():
+        counts = active_job_counts(Database(db_path))
+        active_total = sum(counts.values())
+        if active_total:
+            print(
+                "migration blocked: active jobs must be resolved first "
+                + json.dumps(counts, sort_keys=True),
+                file=sys.stderr,
+            )
+            return 2
+        backup = backup_sqlite_database(
+            db_path,
+            Path(args.backup_dir).expanduser(),
+        )
+    try:
+        queue = JobQueue(db_path, migrate=True)
+    except Exception as exc:
+        rollback_error = ""
+        if backup is not None:
+            try:
+                restore_sqlite_database(db_path, backup["path"])
+            except sqlite3.Error as restore_exc:
+                rollback_error = f"; rollback failed: {restore_exc}"
+        print(f"migration failed: {exc}{rollback_error}", file=sys.stderr)
+        return 1
     with queue.connect() as con:
-        rows = con.execute(
-            "SELECT version,name,checksum,applied_at FROM schema_migrations ORDER BY version"
-        ).fetchall()
+        migrations = migration_history(con)
     print(
         json.dumps(
             {
                 "database": str(queue.path),
-                "migrations": [dict(row) for row in rows],
+                "backup": backup,
+                "migrations": migrations,
             },
             ensure_ascii=False,
         )
@@ -485,7 +524,13 @@ def build_parser() -> argparse.ArgumentParser:
     p.add_argument("--policy", default=None)
     sub = p.add_subparsers(required=True)
     s = sub.add_parser("init-db"); s.set_defaults(func=cmd_init_db)
-    s = sub.add_parser("migrate-db", help="apply pending versioned SQLite migrations"); s.set_defaults(func=cmd_migrate_db)
+    s = sub.add_parser("migrate-db", help="back up the database and apply pending versioned SQLite migrations")
+    s.add_argument(
+        "--backup-dir",
+        default=os.getenv("GITHUB_AGENT_BRIDGE_AUTOUPDATE_BACKUP_DIR", str(DEFAULT_BACKUP_DIR)),
+        help="directory for the required pre-migration SQLite backup",
+    )
+    s.set_defaults(func=cmd_migrate_db)
     s = sub.add_parser("validate-policy", help="validate a policy file against the published schema")
     s.add_argument("--policy", required=True)
     s.set_defaults(func=cmd_validate_policy)

src/github_agent_bridge/persistence/__init__.py

@@ -2,15 +2,21 @@
 
 from .acknowledgements import AcknowledgementClaim, AcknowledgementRepository
 from .commit_statuses import CommitStatusClaim, CommitStatusRepository
-from .database import ClosingConnection, Database, TransactionMode
+from .database import (
+    ClosingConnection,
+    Database,
+    TransactionMode,
+    backup_sqlite_database,
+    restore_sqlite_database,
+)
 from .feedback import (
     FeedbackEvent,
     FeedbackProposal,
     FeedbackRepository,
     FeedbackRule,
 )
 from .ingestion import IngestionRepository, IngestionRequest, IngestionResult
-from .jobs import JobRepository, job_from_row
+from .jobs import ACTIVE_JOB_STATUSES, JobRepository, active_job_counts, job_from_row
 from .mcp_tokens import McpToken, McpTokenCredential, McpTokenRepository
 from .observability import ObservabilityAlert, ObservabilityRepository, ProcessSample
 from .runtime import RuntimeProcess, RuntimeRepository
@@ -21,6 +27,7 @@
 __all__ = [
     "AcknowledgementClaim",
     "AcknowledgementRepository",
+    "ACTIVE_JOB_STATUSES",
     "CommitStatusClaim",
     "CommitStatusRepository",
     "ClosingConnection",
@@ -48,5 +55,8 @@
     "WebhookRepository",
     "WebPushRepository",
     "WebPushSubscription",
+    "active_job_counts",
+    "backup_sqlite_database",
     "job_from_row",
+    "restore_sqlite_database",
 ]

src/github_agent_bridge/persistence/database.py

@@ -2,9 +2,10 @@
 
 import sqlite3
 from contextlib import contextmanager
+from datetime import datetime, timezone
 from enum import Enum
 from pathlib import Path
-from typing import Iterator
+from typing import Any, Iterator
 
 
 class TransactionMode(str, Enum):
@@ -109,3 +110,37 @@ def _configure(self, con: sqlite3.Connection, *, read_only: bool) -> None:
             journal_mode = str(con.execute("PRAGMA journal_mode").fetchone()[0]).lower()
             if journal_mode != "wal":
                 con.execute("PRAGMA journal_mode=WAL")
+
+
+def backup_sqlite_database(
+    path: str | Path,
+    backup_dir: str | Path,
+) -> dict[str, Any]:
+    """Create a consistent online SQLite backup for an operator action."""
+    source_path = Path(path).expanduser()
+    backup_root = Path(backup_dir).expanduser()
+    backup_root.mkdir(parents=True, exist_ok=True)
+    timestamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%S%fZ")
+    backup_path = backup_root / f"{source_path.stem}-{timestamp}.sqlite3"
+    with sqlite3.connect(source_path) as source, sqlite3.connect(backup_path) as target:
+        source.backup(target)
+    return {
+        "path": str(backup_path),
+        "created_at": datetime.now(timezone.utc).isoformat().replace("+00:00", "Z"),
+        "source": str(source_path),
+        "size_bytes": backup_path.stat().st_size,
+    }
+
+
+def restore_sqlite_database(path: str | Path, backup_path: str | Path) -> dict[str, Any]:
+    """Restore a SQLite database from an operator-created backup."""
+    target_path = Path(path).expanduser()
+    source_path = Path(backup_path).expanduser()
+    with sqlite3.connect(source_path) as source, sqlite3.connect(target_path) as target:
+        source.backup(target)
+    return {
+        "restored_at": datetime.now(timezone.utc).isoformat().replace("+00:00", "Z"),
+        "source": str(source_path),
+        "target": str(target_path),
+        "size_bytes": target_path.stat().st_size,
+    }

src/github_agent_bridge/persistence/jobs.py

@@ -10,6 +10,27 @@
 from .runtime import RuntimeRepository
 from .state import StateRepository
 
+ACTIVE_JOB_STATUSES = ("pending", "running", "waiting_approval")
+
+
+def active_job_counts(database: Database) -> dict[str, int]:
+    """Read migration-blocking job counts without constructing a queue."""
+    with database.read_only() as con:
+        jobs_exists = con.execute(
+            "SELECT 1 FROM sqlite_master WHERE type='table' AND name='jobs'"
+        ).fetchone()
+        if jobs_exists is None:
+            return {status: 0 for status in ACTIVE_JOB_STATUSES}
+        placeholders = ",".join("?" for _ in ACTIVE_JOB_STATUSES)
+        rows = con.execute(
+            f"SELECT status,count(*) count FROM jobs "
+            f"WHERE status IN ({placeholders}) GROUP BY status",
+            ACTIVE_JOB_STATUSES,
+        ).fetchall()
+    counts = {status: 0 for status in ACTIVE_JOB_STATUSES}
+    counts.update({str(row["status"]): int(row["count"]) for row in rows})
+    return counts
+
 
 def job_from_row(row: sqlite3.Row | None) -> Job | None:
     """Map one persistence row to the public queue job DTO."""