From 7d36ec5e1fc3c2f15ffa69f6047214947d0ffb34 Mon Sep 17 00:00:00 2001
From: atlas <atlas@hyperhive.local>
Date: Fri, 10 Jul 2026 14:50:39 +0200
Subject: [PATCH 1/3] fix(#2279): versioned DB migrations via schema_versions
 table

Replace the try-and-ignore-duplicate-column approach in apply_migrations
with proper schema versioning using a shared schema_versions table.

## mechanism

New function: db::apply_versioned_migrations(conn, subsystem,
legacy_column, migrations). Tracks the applied-migration count in a
schema_versions table (one row per subsystem key). Only migrations past
the stored version run.

Legacy detection: pre-versioning databases have no schema_versions row.
The legacy_column tuple (table, column) identifies a column that exists
only in a fully-migrated legacy database. If present, all known
migrations are skipped. If absent, migrations start from 0.

## stores migrated

- broker: removes bespoke ensure_message_columns / ensure_reminder_columns.
  Unified into BROKER_MIGRATIONS (v1-v5). Legacy detector: messages.priority
  (added in the last pre-versioning migration).
- approvals: 4 historical migrations (v1-v4). Legacy detector:
  approvals.submitter.
- operator_questions: 3 historical migrations (v1-v3). Legacy detector:
  operator_questions.target.
- scheduled_prompts: 1 historical migration (v1). Legacy detector:
  scheduled_prompts.paused_at_unix.

apply_migrations removed (no callers).

## tests (db.rs)

- fresh_install_runs_all_migrations
- legacy_install_skips_all_migrations
- partial_migration_resumes_from_version
- already_at_latest_is_noop
- multiple_stores_in_same_db
---
 hive-c0re/src/stores/approvals.rs          |  30 +--
 hive-c0re/src/stores/broker.rs             |  93 +++----
 hive-c0re/src/stores/db.rs                 | 290 +++++++++++++++++++--
 hive-c0re/src/stores/operator_questions.rs |  23 +-
 hive-c0re/src/stores/scheduled_prompts.rs  |  12 +-
 5 files changed, 333 insertions(+), 115 deletions(-)

diff --git a/hive-c0re/src/stores/approvals.rs b/hive-c0re/src/stores/approvals.rs
index f9235a2d..37e809ca 100644
--- a/hive-c0re/src/stores/approvals.rs
+++ b/hive-c0re/src/stores/approvals.rs
@@ -24,24 +24,19 @@ CREATE INDEX IF NOT EXISTS idx_approvals_pending
     ON approvals (id) WHERE status = 'pending';
 ";
 
-/// Additive column migrations for pre-existing databases, applied via
-/// `db::apply_migrations` (try-and-ignore-duplicate-column).
+/// Ordered schema migrations tracked in `schema_versions` (key `"approvals"`).
+/// Legacy databases are detected via the `submitter` column — the last
+/// column added before versioning was introduced — and fast-forwarded past
+/// all known migrations. New columns go here as v5, v6, …
 const MIGRATIONS: &[&str] = &[
-    // `kind` (pre-Phase-8 dbs): legacy rows default to `apply_commit`,
-    // which matches their actual semantics.
+    // v1: `kind` (pre-Phase-8 dbs): legacy rows default to `apply_commit`.
     "ALTER TABLE approvals ADD COLUMN kind TEXT NOT NULL DEFAULT 'apply_commit'",
-    // `description`: manager-supplied note shown on the dashboard
-    // approval card at submission time (distinct from `note`, set on
-    // denial/failure).
+    // v2: `description`: manager-supplied note on the dashboard card.
     "ALTER TABLE approvals ADD COLUMN description TEXT",
-    // `fetched_sha`: the canonical sha hive-c0re vouched for at
-    // `request_apply_commit` time. Distinct from `commit_ref`
-    // (manager-supplied, may not even resolve by approve time).
+    // v3: `fetched_sha`: canonical sha hive-c0re resolved at submit time.
     "ALTER TABLE approvals ADD COLUMN fetched_sha TEXT",
-    // `submitter`: the agent that submitted the approval (the
-    // authenticated socket caller); approval-scoped helper events
-    // route to it. Legacy rows are NULL → callers fall back to the
-    // root agent.
+    // v4: `submitter`: authenticated agent that submitted the approval.
+    // Legacy rows are NULL → callers fall back to the root agent.
     "ALTER TABLE approvals ADD COLUMN submitter TEXT",
 ];
 
@@ -54,7 +49,12 @@ impl Approvals {
         let conn = crate::db::open(path, "approvals")?;
         conn.execute_batch(SCHEMA)
             .context("apply approvals schema")?;
-        crate::db::apply_migrations(&conn, "approvals", MIGRATIONS)?;
+        crate::db::apply_versioned_migrations(
+            &conn,
+            "approvals",
+            ("approvals", "submitter"),
+            MIGRATIONS,
+        )?;
         Ok(Self {
             conn: Mutex::new(conn),
         })
diff --git a/hive-c0re/src/stores/broker.rs b/hive-c0re/src/stores/broker.rs
index cfff76b0..b779674b 100644
--- a/hive-c0re/src/stores/broker.rs
+++ b/hive-c0re/src/stores/broker.rs
@@ -156,12 +156,42 @@ pub struct Broker {
     inflight: Mutex<HashMap<String, RecipientInflight>>,
 }
 
+/// Ordered schema migrations for the broker. Tracked in `schema_versions`
+/// under key `"broker"`. Legacy databases (fully migrated via the old
+/// per-column check approach) are detected via the `priority` column on
+/// `messages` — the last column added before versioning.
+const BROKER_MIGRATIONS: &[&str] = &[
+    // v1: acked_at on messages, with backfill so existing delivered rows
+    // are not phantom-requeued on the next open. The multi-statement
+    // batch runs atomically (execute_batch wraps in a transaction).
+    "ALTER TABLE messages ADD COLUMN acked_at INTEGER;\
+     UPDATE messages SET acked_at = delivered_at WHERE delivered_at IS NOT NULL;",
+    // v2: in_reply_to for thread-parent tracking. NULL = root of a thread.
+    "ALTER TABLE messages ADD COLUMN in_reply_to INTEGER",
+    // v3: priority for operator-message fast-path. Rebuild the delivery
+    // index to include priority as a secondary sort key.
+    "ALTER TABLE messages ADD COLUMN priority INTEGER NOT NULL DEFAULT 0;\
+     DROP INDEX IF EXISTS idx_messages_undelivered;\
+     CREATE INDEX idx_messages_undelivered \
+         ON messages (recipient, priority DESC, id) WHERE delivered_at IS NULL",
+    // v4: attempt_count on reminders for the MAX_REMINDER_ATTEMPTS cap.
+    "ALTER TABLE reminders ADD COLUMN attempt_count INTEGER NOT NULL DEFAULT 0",
+    // v5: last_error on reminders — last delivery failure surfaced on the
+    // dashboard so a stuck reminder is visible without digging in logs.
+    "ALTER TABLE reminders ADD COLUMN last_error TEXT",
+];
+
 impl Broker {
     pub fn open(path: &Path) -> Result<Self> {
         let conn = crate::db::open(path, "broker")?;
         conn.execute_batch(SCHEMA).context("apply broker schema")?;
-        ensure_message_columns(&conn).context("migrate messages columns")?;
-        ensure_reminder_columns(&conn).context("migrate reminders columns")?;
+        crate::db::apply_versioned_migrations(
+            &conn,
+            "broker",
+            ("messages", "priority"),
+            BROKER_MIGRATIONS,
+        )
+        .context("broker migrations")?;
         let (events, _) = broadcast::channel(EVENT_CHANNEL);
         Ok(Self {
             conn: Mutex::new(conn),
@@ -1108,65 +1138,6 @@ impl Broker {
 /// back-fills it for every already-delivered row, so the
 /// pre-migration sessions count as "fully handled" and won't be
 /// resurfaced by the first `requeue_inflight` after upgrade.
-fn ensure_message_columns(conn: &Connection) -> Result<()> {
-    let has_acked: bool = conn
-        .prepare("SELECT 1 FROM pragma_table_info('messages') WHERE name = 'acked_at'")?
-        .exists([])?;
-    if !has_acked {
-        conn.execute_batch("ALTER TABLE messages ADD COLUMN acked_at INTEGER;")
-            .context("add messages.acked_at column")?;
-        // Backfill: treat every existing delivered row as acked. The
-        // session it was delivered to is gone, so requeue would just
-        // surface phantom traffic to whatever harness reads next.
-        conn.execute(
-            "UPDATE messages SET acked_at = delivered_at \
-              WHERE delivered_at IS NOT NULL AND acked_at IS NULL",
-            [],
-        )
-        .context("backfill messages.acked_at from delivered_at")?;
-    }
-    let has_reply: bool = conn
-        .prepare("SELECT 1 FROM pragma_table_info('messages') WHERE name = 'in_reply_to'")?
-        .exists([])?;
-    if !has_reply {
-        conn.execute_batch("ALTER TABLE messages ADD COLUMN in_reply_to INTEGER;")
-            .context("add messages.in_reply_to column")?;
-        // No backfill needed — existing messages simply have NULL here,
-        // meaning "root of a new thread", which is correct.
-    }
-    let has_priority: bool = conn
-        .prepare("SELECT 1 FROM pragma_table_info('messages') WHERE name = 'priority'")?
-        .exists([])?;
-    if !has_priority {
-        // Add column + rebuild the index with priority as the second key so
-        // recv_batch can serve operator messages ahead of queued wakes without
-        // a separate sort pass.
-        conn.execute_batch(
-            "ALTER TABLE messages ADD COLUMN priority INTEGER NOT NULL DEFAULT 0;
-             DROP INDEX IF EXISTS idx_messages_undelivered;
-             CREATE INDEX idx_messages_undelivered
-                 ON messages (recipient, priority DESC, id) WHERE delivered_at IS NULL;",
-        )
-        .context("add messages.priority column and update delivery index")?;
-        // No backfill needed — all existing rows are correctly at priority 0.
-    }
-    Ok(())
-}
-
-/// Idempotent reminder-table migrations — plain additive columns, via
-/// `db::apply_migrations`. (The messages-table migration above stays
-/// bespoke: its backfill must run only when `acked_at` was just
-/// created, which try-and-ignore can't express.)
-fn ensure_reminder_columns(conn: &Connection) -> Result<()> {
-    crate::db::apply_migrations(
-        conn,
-        "broker reminders",
-        &[
-            "ALTER TABLE reminders ADD COLUMN attempt_count INTEGER NOT NULL DEFAULT 0",
-            "ALTER TABLE reminders ADD COLUMN last_error TEXT",
-        ],
-    )
-}
 
 #[cfg(test)]
 mod tests {
diff --git a/hive-c0re/src/stores/db.rs b/hive-c0re/src/stores/db.rs
index f483d3d2..886fa96e 100644
--- a/hive-c0re/src/stores/db.rs
+++ b/hive-c0re/src/stores/db.rs
@@ -1,43 +1,114 @@
-//! Shared sqlite connection setup for hive-c0re's host-side stores.
+//! Shared sqlite connection setup and versioned schema migrations for
+//! hive-c0re's host-side stores.
 //!
-//! Several modules keep their own tables — and their own
-//! `Mutex<Connection>` — in the coordinator DB
-//! (`db/broker.sqlite`: broker, approvals, operator questions,
-//! scheduled prompts, agent power) or in a sibling file under the same
-//! `db/` dir (`build_logs.sqlite`, `audit_log.sqlite`). The open dance
-//! is identical everywhere: ensure the parent dir exists, open the
-//! connection, set a busy timeout so concurrent same-process writers
-//! wait each other out instead of surfacing `SQLITE_BUSY`. This helper
-//! owns that dance; schema creation + column migrations stay with each
-//! store (they're per-table concerns).
+//! All stores share one sqlite file (`db/broker.sqlite`) but each owns
+//! its own `Mutex<Connection>`. [`open`] handles the connection open
+//! dance; [`apply_versioned_migrations`] handles schema evolution.
+//!
+//! ## Migration strategy
+//!
+//! Each store calls [`apply_versioned_migrations`] from its `open`. It
+//! tracks the applied count in a `schema_versions` table (one row per
+//! `subsystem` key). Only migrations past the stored version run.
+//!
+//! Legacy databases (written before versioning) are detected via
+//! `legacy_column = (table, column)` — a column that exists only in
+//! a fully-migrated legacy DB. Present → skip all migrations. Absent →
+//! start from 0. See each store's `MIGRATIONS` constant for the list.
 
 use std::path::Path;
 use std::time::Duration;
 
 use anyhow::{Context, Result};
-use rusqlite::Connection;
+use rusqlite::{Connection, OptionalExtension, params};
 
 /// How long a write waits on another connection's lock before erroring.
 /// Generous relative to the stores' tiny transactions — a timeout here
 /// means something is genuinely wedged, not ordinary contention.
 const BUSY_TIMEOUT: Duration = Duration::from_secs(5);
 
-/// Apply additive, idempotent migrations: run each statement and
-/// ignore `duplicate column name` errors (sqlite has no
-/// `ADD COLUMN IF NOT EXISTS`; try-and-ignore is the portable path —
-/// same pattern as hive-ag3nt's `turn_stats`). Any other error
-/// propagates. Fits plain `ALTER TABLE ADD COLUMN` (new columns must
-/// carry a default or tolerate NULL) and `IF NOT EXISTS` index DDL;
-/// migrations with creation-conditional backfills (broker's
-/// `acked_at`) stay bespoke in their store.
-pub fn apply_migrations(conn: &Connection, subsystem: &str, statements: &[&str]) -> Result<()> {
-    for stmt in statements {
-        if let Err(e) = conn.execute_batch(stmt) {
-            if e.to_string().contains("duplicate column name") {
-                continue;
-            }
-            return Err(e).with_context(|| format!("{subsystem} migration failed: {stmt}"));
-        }
+const SCHEMA_VERSIONS_DDL: &str = "
+CREATE TABLE IF NOT EXISTS schema_versions (
+    store   TEXT    PRIMARY KEY,
+    version INTEGER NOT NULL DEFAULT 0
+)";
+
+/// Apply numbered schema migrations for `subsystem`, tracking progress in
+/// the shared `schema_versions` table.
+///
+/// - `subsystem` — unique key for this store in `schema_versions`
+///   (e.g. `"approvals"`, `"broker"`).
+/// - `legacy_column` — `(table_name, column_name)` of a column that exists
+///   in a fully-migrated pre-versioning database. When there is no
+///   `schema_versions` row yet and this column is present, all `migrations`
+///   are skipped (they were previously applied via the old try-and-ignore
+///   approach). When the column is absent, we start at version 0.
+/// - `migrations` — ordered list of SQL batches; each batch runs once, at
+///   the index whose value is the current version.
+///
+/// # Errors
+///
+/// Returns an error if any migration statement fails or the version update
+/// fails. A failed migration leaves the DB at the last successfully
+/// committed version (each step is its own implicit transaction via
+/// `execute_batch`).
+pub fn apply_versioned_migrations(
+    conn: &Connection,
+    subsystem: &str,
+    legacy_column: (&str, &str),
+    migrations: &[&str],
+) -> Result<()> {
+    // Ensure the version-tracking table exists (idempotent).
+    conn.execute_batch(SCHEMA_VERSIONS_DDL)
+        .context("create schema_versions table")?;
+
+    // Look up the stored version for this subsystem.
+    let stored: Option<i64> = conn
+        .query_row(
+            "SELECT version FROM schema_versions WHERE store = ?1",
+            [subsystem],
+            |row| row.get(0),
+        )
+        .optional()
+        .context("read schema_versions")?;
+
+    let version = if let Some(v) = stored {
+        usize::try_from(v.max(0)).unwrap_or(0)
+    } else {
+        // No entry yet. Check the legacy-column to decide where to start.
+        // If it exists, all known migrations were already applied by the
+        // old try-and-ignore path — skip them. If it doesn't exist, start
+        // at 0 (fresh install or partial state).
+        let (legacy_table, legacy_col) = legacy_column;
+        let is_legacy: bool = conn
+            .prepare(&format!(
+                "SELECT 1 FROM pragma_table_info('{legacy_table}') WHERE name = ?1"
+            ))
+            .context("prepare legacy-column probe")?
+            .exists([legacy_col])
+            .context("check legacy column")?;
+        let initial = if is_legacy { migrations.len() } else { 0 };
+        conn.execute(
+            "INSERT INTO schema_versions (store, version) VALUES (?1, ?2)",
+            params![subsystem, i64::try_from(initial).unwrap_or(0)],
+        )
+        .context("insert schema_versions row")?;
+        initial
+    };
+
+    if version >= migrations.len() {
+        return Ok(());
+    }
+
+    for (i, stmt) in migrations.iter().enumerate().skip(version) {
+        conn.execute_batch(stmt)
+            .with_context(|| format!("{subsystem} migration v{} failed: {stmt}", i + 1))?;
+        let next = i64::try_from(i + 1).unwrap_or(i64::MAX);
+        conn.execute(
+            "UPDATE schema_versions SET version = ?1 WHERE store = ?2",
+            params![next, subsystem],
+        )
+        .with_context(|| format!("{subsystem} update schema_versions to v{}", i + 1))?;
     }
     Ok(())
 }
@@ -56,3 +127,166 @@ pub fn open(path: &Path, subsystem: &str) -> Result<Connection> {
         .with_context(|| format!("set {subsystem} busy_timeout"))?;
     Ok(conn)
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::sync::atomic::{AtomicU64, Ordering};
+
+    static TEST_COUNTER: AtomicU64 = AtomicU64::new(0);
+
+    fn open_tmp() -> Connection {
+        let n = TEST_COUNTER.fetch_add(1, Ordering::Relaxed);
+        let path = format!("/tmp/hive-db-test-{}-{}.sqlite", std::process::id(), n);
+        Connection::open(&path).expect("open tmp db")
+    }
+
+    const TEST_SCHEMA: &str = "CREATE TABLE IF NOT EXISTS things (id INTEGER PRIMARY KEY)";
+    const TEST_MIGRATIONS: &[&str] = &[
+        "ALTER TABLE things ADD COLUMN alpha TEXT",
+        "ALTER TABLE things ADD COLUMN beta INTEGER NOT NULL DEFAULT 0",
+        "ALTER TABLE things ADD COLUMN gamma TEXT",
+    ];
+
+    #[test]
+    fn fresh_install_runs_all_migrations() {
+        let conn = open_tmp();
+        conn.execute_batch(TEST_SCHEMA).unwrap();
+        apply_versioned_migrations(&conn, "things", ("things", "gamma"), TEST_MIGRATIONS).unwrap();
+        // All three columns should exist.
+        for col in ["alpha", "beta", "gamma"] {
+            let exists: bool = conn
+                .prepare(&format!(
+                    "SELECT 1 FROM pragma_table_info('things') WHERE name = '{col}'"
+                ))
+                .unwrap()
+                .exists([])
+                .unwrap();
+            assert!(exists, "column {col} missing after fresh migration");
+        }
+        // Version should be 3.
+        let v: i64 = conn
+            .query_row(
+                "SELECT version FROM schema_versions WHERE store = 'things'",
+                [],
+                |r| r.get(0),
+            )
+            .unwrap();
+        assert_eq!(v, 3);
+    }
+
+    #[test]
+    fn legacy_install_skips_all_migrations() {
+        let conn = open_tmp();
+        // Simulate a fully-migrated legacy DB: all columns exist,
+        // but schema_versions does not yet.
+        conn.execute_batch(
+            "CREATE TABLE IF NOT EXISTS things (id INTEGER PRIMARY KEY,
+             alpha TEXT, beta INTEGER NOT NULL DEFAULT 0, gamma TEXT)",
+        )
+        .unwrap();
+        apply_versioned_migrations(&conn, "things", ("things", "gamma"), TEST_MIGRATIONS).unwrap();
+        // Version should be set to migrations.len() (3), not 0.
+        let v: i64 = conn
+            .query_row(
+                "SELECT version FROM schema_versions WHERE store = 'things'",
+                [],
+                |r| r.get(0),
+            )
+            .unwrap();
+        assert_eq!(v, 3, "legacy install should skip to latest version");
+    }
+
+    #[test]
+    fn partial_migration_resumes_from_version() {
+        let conn = open_tmp();
+        // Simulate a DB that has only the first migration applied.
+        conn.execute_batch(
+            "CREATE TABLE IF NOT EXISTS things (id INTEGER PRIMARY KEY, alpha TEXT);
+             CREATE TABLE IF NOT EXISTS schema_versions (store TEXT PRIMARY KEY, version INTEGER NOT NULL DEFAULT 0);
+             INSERT INTO schema_versions (store, version) VALUES ('things', 1)",
+        )
+        .unwrap();
+        apply_versioned_migrations(&conn, "things", ("things", "gamma"), TEST_MIGRATIONS).unwrap();
+        // Only beta and gamma should have been added (alpha already existed).
+        for col in ["alpha", "beta", "gamma"] {
+            let exists: bool = conn
+                .prepare(&format!(
+                    "SELECT 1 FROM pragma_table_info('things') WHERE name = '{col}'"
+                ))
+                .unwrap()
+                .exists([])
+                .unwrap();
+            assert!(
+                exists,
+                "column {col} missing after partial migration resume"
+            );
+        }
+        let v: i64 = conn
+            .query_row(
+                "SELECT version FROM schema_versions WHERE store = 'things'",
+                [],
+                |r| r.get(0),
+            )
+            .unwrap();
+        assert_eq!(v, 3);
+    }
+
+    #[test]
+    fn already_at_latest_is_noop() {
+        let conn = open_tmp();
+        conn.execute_batch(TEST_SCHEMA).unwrap();
+        apply_versioned_migrations(&conn, "things", ("things", "gamma"), TEST_MIGRATIONS).unwrap();
+        // Second call is a no-op.
+        apply_versioned_migrations(&conn, "things", ("things", "gamma"), TEST_MIGRATIONS).unwrap();
+        let v: i64 = conn
+            .query_row(
+                "SELECT version FROM schema_versions WHERE store = 'things'",
+                [],
+                |r| r.get(0),
+            )
+            .unwrap();
+        assert_eq!(v, 3);
+    }
+
+    #[test]
+    fn multiple_stores_in_same_db() {
+        let conn = open_tmp();
+        conn.execute_batch(
+            "CREATE TABLE IF NOT EXISTS a (id INTEGER PRIMARY KEY);
+             CREATE TABLE IF NOT EXISTS b (id INTEGER PRIMARY KEY)",
+        )
+        .unwrap();
+        apply_versioned_migrations(
+            &conn,
+            "store_a",
+            ("a", "col1"),
+            &["ALTER TABLE a ADD COLUMN col1 TEXT"],
+        )
+        .unwrap();
+        apply_versioned_migrations(
+            &conn,
+            "store_b",
+            ("b", "colx"),
+            &["ALTER TABLE b ADD COLUMN colx INTEGER NOT NULL DEFAULT 0"],
+        )
+        .unwrap();
+        // Each store tracks independently.
+        let va: i64 = conn
+            .query_row(
+                "SELECT version FROM schema_versions WHERE store = 'store_a'",
+                [],
+                |r| r.get(0),
+            )
+            .unwrap();
+        let vb: i64 = conn
+            .query_row(
+                "SELECT version FROM schema_versions WHERE store = 'store_b'",
+                [],
+                |r| r.get(0),
+            )
+            .unwrap();
+        assert_eq!(va, 1);
+        assert_eq!(vb, 1);
+    }
+}
diff --git a/hive-c0re/src/stores/operator_questions.rs b/hive-c0re/src/stores/operator_questions.rs
index 2a99699f..87ac6214 100644
--- a/hive-c0re/src/stores/operator_questions.rs
+++ b/hive-c0re/src/stores/operator_questions.rs
@@ -33,16 +33,18 @@ CREATE INDEX IF NOT EXISTS idx_operator_questions_pending
     ON operator_questions (id) WHERE answered_at IS NULL;
 ";
 
-/// Additive column migrations for pre-existing databases, applied via
-/// `db::apply_migrations` (try-and-ignore-duplicate-column).
+/// Ordered schema migrations tracked in `schema_versions` (key
+/// `"operator_questions"`). Legacy databases are detected via the `target`
+/// column — the last column added before versioning — and fast-forwarded
+/// past all known migrations.
 const MIGRATIONS: &[&str] = &[
+    // v1: `multi` — checkbox-style multi-option questions.
     "ALTER TABLE operator_questions ADD COLUMN multi INTEGER NOT NULL DEFAULT 0",
+    // v2: `deadline_at` — optional TTL after which the watchdog auto-resolves.
     "ALTER TABLE operator_questions ADD COLUMN deadline_at INTEGER",
-    // `target` = recipient of the question. NULL = operator
-    // (back-compat default for rows written before agent-to-agent
-    // questions existed); a non-null agent name = peer-to-peer
-    // question. Dashboard's `pending()` filters on `target IS NULL`
-    // so peer questions never leak into the operator's queue.
+    // v3: `target` — recipient of the question. NULL = operator (back-compat
+    // default); non-null = peer-to-peer question. Dashboard's `pending()`
+    // filters on `target IS NULL` so peer questions never leak to the operator.
     "ALTER TABLE operator_questions ADD COLUMN target TEXT",
 ];
 
@@ -77,7 +79,12 @@ impl OperatorQuestions {
         let conn = crate::db::open(path, "operator_questions")?;
         conn.execute_batch(SCHEMA)
             .context("apply operator_questions schema")?;
-        crate::db::apply_migrations(&conn, "operator_questions", MIGRATIONS)?;
+        crate::db::apply_versioned_migrations(
+            &conn,
+            "operator_questions",
+            ("operator_questions", "target"),
+            MIGRATIONS,
+        )?;
         Ok(Self {
             conn: Mutex::new(conn),
         })
diff --git a/hive-c0re/src/stores/scheduled_prompts.rs b/hive-c0re/src/stores/scheduled_prompts.rs
index 7aaaff1f..7d745aa9 100644
--- a/hive-c0re/src/stores/scheduled_prompts.rs
+++ b/hive-c0re/src/stores/scheduled_prompts.rs
@@ -186,11 +186,17 @@ impl ScheduledPrompts {
             .context("enable foreign keys")?;
         conn.execute_batch(SCHEMA)
             .context("apply scheduled_prompts schema")?;
-        // Migration: add paused_at_unix to existing databases.
-        crate::db::apply_migrations(
+        // Versioned migration: add paused_at_unix. Legacy databases are
+        // detected via the presence of this column (it was the only
+        // migration before versioning was introduced).
+        crate::db::apply_versioned_migrations(
             &conn,
             "scheduled_prompts",
-            &["ALTER TABLE scheduled_prompts ADD COLUMN paused_at_unix INTEGER"],
+            ("scheduled_prompts", "paused_at_unix"),
+            &[
+                // v1: add paused_at_unix for per-schedule pause support.
+                "ALTER TABLE scheduled_prompts ADD COLUMN paused_at_unix INTEGER",
+            ],
         )?;
         // Migration: recreate the due-rows index to also exclude paused
         // rows. `CREATE INDEX IF NOT EXISTS` won't update an existing

From c47faf0af96151baa748953bf330b98fc8063943 Mon Sep 17 00:00:00 2001
From: atlas <atlas@hyperhive.local>
Date: Fri, 10 Jul 2026 16:25:02 +0200
Subject: [PATCH 2/3] fix: idempotent migrations (ADD COLUMN IF NOT EXISTS),
 stale doc, safety note

---
 hive-c0re/src/stores/approvals.rs          |  9 ++--
 hive-c0re/src/stores/broker.rs             | 36 ++++++++-------
 hive-c0re/src/stores/db.rs                 | 51 ++++++++++++++++++++--
 hive-c0re/src/stores/operator_questions.rs |  7 +--
 hive-c0re/src/stores/scheduled_prompts.rs  |  3 +-
 5 files changed, 80 insertions(+), 26 deletions(-)

diff --git a/hive-c0re/src/stores/approvals.rs b/hive-c0re/src/stores/approvals.rs
index 37e809ca..a8f39633 100644
--- a/hive-c0re/src/stores/approvals.rs
+++ b/hive-c0re/src/stores/approvals.rs
@@ -30,14 +30,15 @@ CREATE INDEX IF NOT EXISTS idx_approvals_pending
 /// all known migrations. New columns go here as v5, v6, …
 const MIGRATIONS: &[&str] = &[
     // v1: `kind` (pre-Phase-8 dbs): legacy rows default to `apply_commit`.
-    "ALTER TABLE approvals ADD COLUMN kind TEXT NOT NULL DEFAULT 'apply_commit'",
+    "ALTER TABLE approvals ADD COLUMN IF NOT EXISTS \
+     kind TEXT NOT NULL DEFAULT 'apply_commit'",
     // v2: `description`: manager-supplied note on the dashboard card.
-    "ALTER TABLE approvals ADD COLUMN description TEXT",
+    "ALTER TABLE approvals ADD COLUMN IF NOT EXISTS description TEXT",
     // v3: `fetched_sha`: canonical sha hive-c0re resolved at submit time.
-    "ALTER TABLE approvals ADD COLUMN fetched_sha TEXT",
+    "ALTER TABLE approvals ADD COLUMN IF NOT EXISTS fetched_sha TEXT",
     // v4: `submitter`: authenticated agent that submitted the approval.
     // Legacy rows are NULL → callers fall back to the root agent.
-    "ALTER TABLE approvals ADD COLUMN submitter TEXT",
+    "ALTER TABLE approvals ADD COLUMN IF NOT EXISTS submitter TEXT",
 ];
 
 pub struct Approvals {
diff --git a/hive-c0re/src/stores/broker.rs b/hive-c0re/src/stores/broker.rs
index b779674b..c614008e 100644
--- a/hive-c0re/src/stores/broker.rs
+++ b/hive-c0re/src/stores/broker.rs
@@ -162,23 +162,34 @@ pub struct Broker {
 /// `messages` — the last column added before versioning.
 const BROKER_MIGRATIONS: &[&str] = &[
     // v1: acked_at on messages, with backfill so existing delivered rows
-    // are not phantom-requeued on the next open. The multi-statement
-    // batch runs atomically (execute_batch wraps in a transaction).
-    "ALTER TABLE messages ADD COLUMN acked_at INTEGER;\
-     UPDATE messages SET acked_at = delivered_at WHERE delivered_at IS NOT NULL;",
+    // are not phantom-requeued on the next open.
+    //
+    // IF NOT EXISTS: safe to run on a partially-migrated legacy DB where
+    // acked_at already exists but priority (the legacy marker) does not.
+    // The BEGIN/COMMIT block makes ALTER + UPDATE atomic — if the UPDATE
+    // fails the version counter is not bumped and the whole step retries.
+    "BEGIN;\
+     ALTER TABLE messages ADD COLUMN IF NOT EXISTS acked_at INTEGER;\
+     UPDATE messages SET acked_at = delivered_at \
+         WHERE delivered_at IS NOT NULL AND acked_at IS NULL;\
+     COMMIT;",
     // v2: in_reply_to for thread-parent tracking. NULL = root of a thread.
-    "ALTER TABLE messages ADD COLUMN in_reply_to INTEGER",
+    "ALTER TABLE messages ADD COLUMN IF NOT EXISTS in_reply_to INTEGER",
     // v3: priority for operator-message fast-path. Rebuild the delivery
     // index to include priority as a secondary sort key.
-    "ALTER TABLE messages ADD COLUMN priority INTEGER NOT NULL DEFAULT 0;\
+    "BEGIN;\
+     ALTER TABLE messages ADD COLUMN IF NOT EXISTS \
+         priority INTEGER NOT NULL DEFAULT 0;\
      DROP INDEX IF EXISTS idx_messages_undelivered;\
-     CREATE INDEX idx_messages_undelivered \
-         ON messages (recipient, priority DESC, id) WHERE delivered_at IS NULL",
+     CREATE INDEX IF NOT EXISTS idx_messages_undelivered \
+         ON messages (recipient, priority DESC, id) WHERE delivered_at IS NULL;\
+     COMMIT;",
     // v4: attempt_count on reminders for the MAX_REMINDER_ATTEMPTS cap.
-    "ALTER TABLE reminders ADD COLUMN attempt_count INTEGER NOT NULL DEFAULT 0",
+    "ALTER TABLE reminders ADD COLUMN IF NOT EXISTS \
+         attempt_count INTEGER NOT NULL DEFAULT 0",
     // v5: last_error on reminders — last delivery failure surfaced on the
     // dashboard so a stuck reminder is visible without digging in logs.
-    "ALTER TABLE reminders ADD COLUMN last_error TEXT",
+    "ALTER TABLE reminders ADD COLUMN IF NOT EXISTS last_error TEXT",
 ];
 
 impl Broker {
@@ -1134,11 +1145,6 @@ impl Broker {
     }
 }
 
-/// Idempotent messages-table migrations. Adds `acked_at` and
-/// back-fills it for every already-delivered row, so the
-/// pre-migration sessions count as "fully handled" and won't be
-/// resurfaced by the first `requeue_inflight` after upgrade.
-
 #[cfg(test)]
 mod tests {
     use super::*;
diff --git a/hive-c0re/src/stores/db.rs b/hive-c0re/src/stores/db.rs
index 886fa96e..c3810a0c 100644
--- a/hive-c0re/src/stores/db.rs
+++ b/hive-c0re/src/stores/db.rs
@@ -80,6 +80,11 @@ pub fn apply_versioned_migrations(
         // old try-and-ignore path — skip them. If it doesn't exist, start
         // at 0 (fresh install or partial state).
         let (legacy_table, legacy_col) = legacy_column;
+        // SAFETY: `legacy_table` is always a static table-name literal
+        // supplied by each store's `open` (e.g. `"messages"`). Dynamic
+        // or untrusted table names must not be passed here — if that
+        // ever changes, switch to `pragma_table_info(?1)` with two
+        // bound parameters.
         let is_legacy: bool = conn
             .prepare(&format!(
                 "SELECT 1 FROM pragma_table_info('{legacy_table}') WHERE name = ?1"
@@ -143,9 +148,9 @@ mod tests {
 
     const TEST_SCHEMA: &str = "CREATE TABLE IF NOT EXISTS things (id INTEGER PRIMARY KEY)";
     const TEST_MIGRATIONS: &[&str] = &[
-        "ALTER TABLE things ADD COLUMN alpha TEXT",
-        "ALTER TABLE things ADD COLUMN beta INTEGER NOT NULL DEFAULT 0",
-        "ALTER TABLE things ADD COLUMN gamma TEXT",
+        "ALTER TABLE things ADD COLUMN IF NOT EXISTS alpha TEXT",
+        "ALTER TABLE things ADD COLUMN IF NOT EXISTS beta INTEGER NOT NULL DEFAULT 0",
+        "ALTER TABLE things ADD COLUMN IF NOT EXISTS gamma TEXT",
     ];
 
     #[test]
@@ -289,4 +294,44 @@ mod tests {
         assert_eq!(va, 1);
         assert_eq!(vb, 1);
     }
+
+    /// A partially-migrated legacy database: some columns from the old
+    /// try-and-ignore path exist, but the legacy marker column (used to
+    /// fast-forward fresh schema_versions entries to `len`) is absent.
+    /// With idempotent `ADD COLUMN IF NOT EXISTS` migrations this must
+    /// succeed — the already-present columns are skipped and only the
+    /// missing ones are added.
+    #[test]
+    fn partial_legacy_completes_without_error() {
+        let conn = open_tmp();
+        // Simulate a DB that has `alpha` (v1) but not `beta` (v2) or
+        // `gamma` (v3). No schema_versions row exists yet, and `gamma`
+        // (the legacy marker) is absent → the function cannot fast-forward
+        // and must start from v0 with idempotent migrations.
+        conn.execute_batch(
+            "CREATE TABLE IF NOT EXISTS things (id INTEGER PRIMARY KEY, alpha TEXT)",
+        )
+        .unwrap();
+        apply_versioned_migrations(&conn, "things", ("things", "gamma"), TEST_MIGRATIONS).unwrap();
+        // All three columns must now exist.
+        for col in ["alpha", "beta", "gamma"] {
+            let exists: bool = conn
+                .prepare(&format!(
+                    "SELECT 1 FROM pragma_table_info('things') WHERE name = '{col}'"
+                ))
+                .unwrap()
+                .exists([])
+                .unwrap();
+            assert!(exists, "column {col} missing after partial-legacy migration");
+        }
+        // Version must be at the latest.
+        let v: i64 = conn
+            .query_row(
+                "SELECT version FROM schema_versions WHERE store = 'things'",
+                [],
+                |r| r.get(0),
+            )
+            .unwrap();
+        assert_eq!(v, 3, "must reach latest version after completing partial legacy");
+    }
 }
diff --git a/hive-c0re/src/stores/operator_questions.rs b/hive-c0re/src/stores/operator_questions.rs
index 87ac6214..4f352020 100644
--- a/hive-c0re/src/stores/operator_questions.rs
+++ b/hive-c0re/src/stores/operator_questions.rs
@@ -39,13 +39,14 @@ CREATE INDEX IF NOT EXISTS idx_operator_questions_pending
 /// past all known migrations.
 const MIGRATIONS: &[&str] = &[
     // v1: `multi` — checkbox-style multi-option questions.
-    "ALTER TABLE operator_questions ADD COLUMN multi INTEGER NOT NULL DEFAULT 0",
+    "ALTER TABLE operator_questions ADD COLUMN IF NOT EXISTS \
+     multi INTEGER NOT NULL DEFAULT 0",
     // v2: `deadline_at` — optional TTL after which the watchdog auto-resolves.
-    "ALTER TABLE operator_questions ADD COLUMN deadline_at INTEGER",
+    "ALTER TABLE operator_questions ADD COLUMN IF NOT EXISTS deadline_at INTEGER",
     // v3: `target` — recipient of the question. NULL = operator (back-compat
     // default); non-null = peer-to-peer question. Dashboard's `pending()`
     // filters on `target IS NULL` so peer questions never leak to the operator.
-    "ALTER TABLE operator_questions ADD COLUMN target TEXT",
+    "ALTER TABLE operator_questions ADD COLUMN IF NOT EXISTS target TEXT",
 ];
 
 #[derive(Debug, Clone, Serialize)]
diff --git a/hive-c0re/src/stores/scheduled_prompts.rs b/hive-c0re/src/stores/scheduled_prompts.rs
index 7d745aa9..4ffa132e 100644
--- a/hive-c0re/src/stores/scheduled_prompts.rs
+++ b/hive-c0re/src/stores/scheduled_prompts.rs
@@ -195,7 +195,8 @@ impl ScheduledPrompts {
             ("scheduled_prompts", "paused_at_unix"),
             &[
                 // v1: add paused_at_unix for per-schedule pause support.
-                "ALTER TABLE scheduled_prompts ADD COLUMN paused_at_unix INTEGER",
+                "ALTER TABLE scheduled_prompts \
+                 ADD COLUMN IF NOT EXISTS paused_at_unix INTEGER",
             ],
         )?;
         // Migration: recreate the due-rows index to also exclude paused

From dfbf198ef3d9884b29b9d8b403297e591292e359 Mon Sep 17 00:00:00 2001
From: damocles <damocles@hyperhive.local>
Date: Fri, 10 Jul 2026 16:51:50 +0200
Subject: [PATCH 3/3] fix(#2279): idempotent column-guard migrations, drop
 invalid ADD COLUMN IF NOT EXISTS

---
 hive-c0re/src/stores/approvals.rs          |  37 +++--
 hive-c0re/src/stores/broker.rs             |  60 +++++---
 hive-c0re/src/stores/db.rs                 | 171 +++++++++++++--------
 hive-c0re/src/stores/operator_questions.rs |  26 ++--
 hive-c0re/src/stores/scheduled_prompts.rs  |  15 +-
 5 files changed, 188 insertions(+), 121 deletions(-)

diff --git a/hive-c0re/src/stores/approvals.rs b/hive-c0re/src/stores/approvals.rs
index a8f39633..338c8f13 100644
--- a/hive-c0re/src/stores/approvals.rs
+++ b/hive-c0re/src/stores/approvals.rs
@@ -10,6 +10,8 @@ use hive_sh4re::wire_time::now_unix;
 use hive_sh4re::{Approval, ApprovalKind, ApprovalStatus};
 use rusqlite::{Connection, OptionalExtension, params};
 
+use crate::db::Migration;
+
 const SCHEMA: &str = r"
 CREATE TABLE IF NOT EXISTS approvals (
     id           INTEGER PRIMARY KEY AUTOINCREMENT,
@@ -25,20 +27,32 @@ CREATE INDEX IF NOT EXISTS idx_approvals_pending
 ";
 
 /// Ordered schema migrations tracked in `schema_versions` (key `"approvals"`).
-/// Legacy databases are detected via the `submitter` column — the last
-/// column added before versioning was introduced — and fast-forwarded past
-/// all known migrations. New columns go here as v5, v6, …
-const MIGRATIONS: &[&str] = &[
+/// Each migration declares the column it adds, so a legacy DB (fully or
+/// partially migrated before versioning) converges by skipping the migrations
+/// whose column already exists. New columns go here as v5, v6, …
+const MIGRATIONS: &[Migration] = &[
     // v1: `kind` (pre-Phase-8 dbs): legacy rows default to `apply_commit`.
-    "ALTER TABLE approvals ADD COLUMN IF NOT EXISTS \
+    Migration {
+        sql: "ALTER TABLE approvals ADD COLUMN \
      kind TEXT NOT NULL DEFAULT 'apply_commit'",
+        adds_column: Some(("approvals", "kind")),
+    },
     // v2: `description`: manager-supplied note on the dashboard card.
-    "ALTER TABLE approvals ADD COLUMN IF NOT EXISTS description TEXT",
+    Migration {
+        sql: "ALTER TABLE approvals ADD COLUMN description TEXT",
+        adds_column: Some(("approvals", "description")),
+    },
     // v3: `fetched_sha`: canonical sha hive-c0re resolved at submit time.
-    "ALTER TABLE approvals ADD COLUMN IF NOT EXISTS fetched_sha TEXT",
+    Migration {
+        sql: "ALTER TABLE approvals ADD COLUMN fetched_sha TEXT",
+        adds_column: Some(("approvals", "fetched_sha")),
+    },
     // v4: `submitter`: authenticated agent that submitted the approval.
     // Legacy rows are NULL → callers fall back to the root agent.
-    "ALTER TABLE approvals ADD COLUMN IF NOT EXISTS submitter TEXT",
+    Migration {
+        sql: "ALTER TABLE approvals ADD COLUMN submitter TEXT",
+        adds_column: Some(("approvals", "submitter")),
+    },
 ];
 
 pub struct Approvals {
@@ -50,12 +64,7 @@ impl Approvals {
         let conn = crate::db::open(path, "approvals")?;
         conn.execute_batch(SCHEMA)
             .context("apply approvals schema")?;
-        crate::db::apply_versioned_migrations(
-            &conn,
-            "approvals",
-            ("approvals", "submitter"),
-            MIGRATIONS,
-        )?;
+        crate::db::apply_versioned_migrations(&conn, "approvals", MIGRATIONS)?;
         Ok(Self {
             conn: Mutex::new(conn),
         })
diff --git a/hive-c0re/src/stores/broker.rs b/hive-c0re/src/stores/broker.rs
index c614008e..204d1b1f 100644
--- a/hive-c0re/src/stores/broker.rs
+++ b/hive-c0re/src/stores/broker.rs
@@ -10,6 +10,8 @@ use anyhow::{Context, Result};
 use chrono::{DateTime, Utc};
 use hive_sh4re::wire_time::now_unix;
 use hive_sh4re::{InboxRow, Message};
+
+use crate::db::Migration;
 use rusqlite::{Connection, OptionalExtension, params};
 use serde::Serialize;
 use tokio::sync::broadcast;
@@ -157,52 +159,60 @@ pub struct Broker {
 }
 
 /// Ordered schema migrations for the broker. Tracked in `schema_versions`
-/// under key `"broker"`. Legacy databases (fully migrated via the old
-/// per-column check approach) are detected via the `priority` column on
-/// `messages` — the last column added before versioning.
-const BROKER_MIGRATIONS: &[&str] = &[
+/// under key `"broker"`. Each migration declares the column it adds, so a
+/// legacy DB (fully or partially migrated via the old per-column approach)
+/// is converged by skipping the migrations whose column already exists.
+const BROKER_MIGRATIONS: &[Migration] = &[
     // v1: acked_at on messages, with backfill so existing delivered rows
-    // are not phantom-requeued on the next open.
-    //
-    // IF NOT EXISTS: safe to run on a partially-migrated legacy DB where
-    // acked_at already exists but priority (the legacy marker) does not.
-    // The BEGIN/COMMIT block makes ALTER + UPDATE atomic — if the UPDATE
-    // fails the version counter is not bumped and the whole step retries.
-    "BEGIN;\
-     ALTER TABLE messages ADD COLUMN IF NOT EXISTS acked_at INTEGER;\
+    // are not phantom-requeued on the next open. The BEGIN/COMMIT block
+    // makes ALTER + UPDATE atomic — either both land or neither, so the
+    // guard column (acked_at) faithfully marks the whole step as done.
+    Migration {
+        sql: "BEGIN;\
+     ALTER TABLE messages ADD COLUMN acked_at INTEGER;\
      UPDATE messages SET acked_at = delivered_at \
          WHERE delivered_at IS NOT NULL AND acked_at IS NULL;\
      COMMIT;",
+        adds_column: Some(("messages", "acked_at")),
+    },
     // v2: in_reply_to for thread-parent tracking. NULL = root of a thread.
-    "ALTER TABLE messages ADD COLUMN IF NOT EXISTS in_reply_to INTEGER",
+    Migration {
+        sql: "ALTER TABLE messages ADD COLUMN in_reply_to INTEGER",
+        adds_column: Some(("messages", "in_reply_to")),
+    },
     // v3: priority for operator-message fast-path. Rebuild the delivery
-    // index to include priority as a secondary sort key.
-    "BEGIN;\
-     ALTER TABLE messages ADD COLUMN IF NOT EXISTS \
+    // index to include priority as a secondary sort key. Atomic BEGIN/COMMIT
+    // so priority existing implies the index rebuild also committed.
+    Migration {
+        sql: "BEGIN;\
+     ALTER TABLE messages ADD COLUMN \
          priority INTEGER NOT NULL DEFAULT 0;\
      DROP INDEX IF EXISTS idx_messages_undelivered;\
      CREATE INDEX IF NOT EXISTS idx_messages_undelivered \
          ON messages (recipient, priority DESC, id) WHERE delivered_at IS NULL;\
      COMMIT;",
+        adds_column: Some(("messages", "priority")),
+    },
     // v4: attempt_count on reminders for the MAX_REMINDER_ATTEMPTS cap.
-    "ALTER TABLE reminders ADD COLUMN IF NOT EXISTS \
+    Migration {
+        sql: "ALTER TABLE reminders ADD COLUMN \
          attempt_count INTEGER NOT NULL DEFAULT 0",
+        adds_column: Some(("reminders", "attempt_count")),
+    },
     // v5: last_error on reminders — last delivery failure surfaced on the
     // dashboard so a stuck reminder is visible without digging in logs.
-    "ALTER TABLE reminders ADD COLUMN IF NOT EXISTS last_error TEXT",
+    Migration {
+        sql: "ALTER TABLE reminders ADD COLUMN last_error TEXT",
+        adds_column: Some(("reminders", "last_error")),
+    },
 ];
 
 impl Broker {
     pub fn open(path: &Path) -> Result<Self> {
         let conn = crate::db::open(path, "broker")?;
         conn.execute_batch(SCHEMA).context("apply broker schema")?;
-        crate::db::apply_versioned_migrations(
-            &conn,
-            "broker",
-            ("messages", "priority"),
-            BROKER_MIGRATIONS,
-        )
-        .context("broker migrations")?;
+        crate::db::apply_versioned_migrations(&conn, "broker", BROKER_MIGRATIONS)
+            .context("broker migrations")?;
         let (events, _) = broadcast::channel(EVENT_CHANNEL);
         Ok(Self {
             conn: Mutex::new(conn),
diff --git a/hive-c0re/src/stores/db.rs b/hive-c0re/src/stores/db.rs
index c3810a0c..9980a8d9 100644
--- a/hive-c0re/src/stores/db.rs
+++ b/hive-c0re/src/stores/db.rs
@@ -11,10 +11,12 @@
 //! tracks the applied count in a `schema_versions` table (one row per
 //! `subsystem` key). Only migrations past the stored version run.
 //!
-//! Legacy databases (written before versioning) are detected via
-//! `legacy_column = (table, column)` — a column that exists only in
-//! a fully-migrated legacy DB. Present → skip all migrations. Absent →
-//! start from 0. See each store's `MIGRATIONS` constant for the list.
+//! Legacy databases (written before versioning) need no special marker:
+//! each column-adding migration carries the `(table, column)` it
+//! introduces, so re-walking a pre-versioning DB from version 0 simply
+//! skips the migrations whose column already exists — converging a fully-
+//! or partially-migrated legacy DB without ever re-running an `ADD COLUMN`
+//! against an existing column. See each store's `MIGRATIONS` constant.
 
 use std::path::Path;
 use std::time::Duration;
@@ -33,36 +35,57 @@ CREATE TABLE IF NOT EXISTS schema_versions (
     version INTEGER NOT NULL DEFAULT 0
 )";
 
+/// One numbered schema migration.
+///
+/// `sql` runs once, at the index whose value is the current stored version.
+/// `adds_column`, when set, is the `(table, column)` this migration
+/// introduces: if that column already exists — a pre-versioning DB migrated
+/// by the old try-and-ignore path, or a partially-migrated one — `sql` is
+/// skipped and only the version counter advances. This makes column-adding
+/// migrations idempotent *without* `SQLite`'s unsupported `ALTER TABLE … ADD
+/// COLUMN IF NOT EXISTS`, so re-walking a legacy DB from version 0 never
+/// re-runs an `ADD COLUMN` against a column that is already there (which
+/// would fail with "duplicate column name" and wedge startup permanently).
+///
+/// Multi-statement migrations wrap their `ALTER` + backfill / index rebuild
+/// in an explicit `BEGIN; … COMMIT;` batch, so "the added column exists" is a
+/// faithful proxy for "this whole migration committed".
+pub struct Migration {
+    /// SQL for this version — a single statement or a `BEGIN; … COMMIT;` batch.
+    pub sql: &'static str,
+    /// `(table, column)` this migration adds, if any. Present → the migration
+    /// is skipped when the column already exists (idempotent replay).
+    pub adds_column: Option<(&'static str, &'static str)>,
+}
+
 /// Apply numbered schema migrations for `subsystem`, tracking progress in
 /// the shared `schema_versions` table.
 ///
 /// - `subsystem` — unique key for this store in `schema_versions`
 ///   (e.g. `"approvals"`, `"broker"`).
-/// - `legacy_column` — `(table_name, column_name)` of a column that exists
-///   in a fully-migrated pre-versioning database. When there is no
-///   `schema_versions` row yet and this column is present, all `migrations`
-///   are skipped (they were previously applied via the old try-and-ignore
-///   approach). When the column is absent, we start at version 0.
-/// - `migrations` — ordered list of SQL batches; each batch runs once, at
-///   the index whose value is the current version.
+/// - `migrations` — ordered list of [`Migration`]s; migration `i` runs when
+///   the stored version is `i`. A pre-versioning DB (no `schema_versions`
+///   row) starts at version 0 and re-walks every migration, but each
+///   column-adding migration is skipped when its `adds_column` is already
+///   present — so a fully- or partially-migrated legacy DB converges without
+///   re-running an `ADD COLUMN` against an existing column.
 ///
 /// # Errors
 ///
 /// Returns an error if any migration statement fails or the version update
-/// fails. A failed migration leaves the DB at the last successfully
-/// committed version (each step is its own implicit transaction via
-/// `execute_batch`).
+/// fails. Each migration + its version bump is a single `execute_batch`
+/// (multi-statement migrations wrap themselves in `BEGIN; … COMMIT;`), so a
+/// failure leaves the DB at the last successfully committed version.
 pub fn apply_versioned_migrations(
     conn: &Connection,
     subsystem: &str,
-    legacy_column: (&str, &str),
-    migrations: &[&str],
+    migrations: &[Migration],
 ) -> Result<()> {
     // Ensure the version-tracking table exists (idempotent).
     conn.execute_batch(SCHEMA_VERSIONS_DDL)
         .context("create schema_versions table")?;
 
-    // Look up the stored version for this subsystem.
+    // Current version for this subsystem — 0 (and a fresh row) when absent.
     let stored: Option<i64> = conn
         .query_row(
             "SELECT version FROM schema_versions WHERE store = ?1",
@@ -71,43 +94,32 @@ pub fn apply_versioned_migrations(
         )
         .optional()
         .context("read schema_versions")?;
-
     let version = if let Some(v) = stored {
         usize::try_from(v.max(0)).unwrap_or(0)
     } else {
-        // No entry yet. Check the legacy-column to decide where to start.
-        // If it exists, all known migrations were already applied by the
-        // old try-and-ignore path — skip them. If it doesn't exist, start
-        // at 0 (fresh install or partial state).
-        let (legacy_table, legacy_col) = legacy_column;
-        // SAFETY: `legacy_table` is always a static table-name literal
-        // supplied by each store's `open` (e.g. `"messages"`). Dynamic
-        // or untrusted table names must not be passed here — if that
-        // ever changes, switch to `pragma_table_info(?1)` with two
-        // bound parameters.
-        let is_legacy: bool = conn
-            .prepare(&format!(
-                "SELECT 1 FROM pragma_table_info('{legacy_table}') WHERE name = ?1"
-            ))
-            .context("prepare legacy-column probe")?
-            .exists([legacy_col])
-            .context("check legacy column")?;
-        let initial = if is_legacy { migrations.len() } else { 0 };
         conn.execute(
-            "INSERT INTO schema_versions (store, version) VALUES (?1, ?2)",
-            params![subsystem, i64::try_from(initial).unwrap_or(0)],
+            "INSERT INTO schema_versions (store, version) VALUES (?1, 0)",
+            [subsystem],
         )
         .context("insert schema_versions row")?;
-        initial
+        0
     };
 
     if version >= migrations.len() {
         return Ok(());
     }
 
-    for (i, stmt) in migrations.iter().enumerate().skip(version) {
-        conn.execute_batch(stmt)
-            .with_context(|| format!("{subsystem} migration v{} failed: {stmt}", i + 1))?;
+    for (i, mig) in migrations.iter().enumerate().skip(version) {
+        // Skip a column-adding migration whose column is already present:
+        // a legacy DB migrated by the old path, or a resumed partial one.
+        let already_applied = match mig.adds_column {
+            Some((table, column)) => column_exists(conn, table, column)?,
+            None => false,
+        };
+        if !already_applied {
+            conn.execute_batch(mig.sql)
+                .with_context(|| format!("{subsystem} migration v{} failed: {}", i + 1, mig.sql))?;
+        }
         let next = i64::try_from(i + 1).unwrap_or(i64::MAX);
         conn.execute(
             "UPDATE schema_versions SET version = ?1 WHERE store = ?2",
@@ -118,6 +130,17 @@ pub fn apply_versioned_migrations(
     Ok(())
 }
 
+/// Whether `column` exists on `table` in this connection's schema. Both
+/// arguments are bound parameters — `pragma_table_info(?1)` accepts the table
+/// name as a bound value — so this is not a SQL-injection surface even if a
+/// caller ever passes a non-static name.
+fn column_exists(conn: &Connection, table: &str, column: &str) -> Result<bool> {
+    conn.prepare("SELECT 1 FROM pragma_table_info(?1) WHERE name = ?2")
+        .context("prepare column probe")?
+        .exists(params![table, column])
+        .context("check column existence")
+}
+
 /// Open a connection to the sqlite file at `path`, creating the parent
 /// directory if needed. `subsystem` labels error contexts (`"broker"`,
 /// `"approvals"`, …).
@@ -147,17 +170,26 @@ mod tests {
     }
 
     const TEST_SCHEMA: &str = "CREATE TABLE IF NOT EXISTS things (id INTEGER PRIMARY KEY)";
-    const TEST_MIGRATIONS: &[&str] = &[
-        "ALTER TABLE things ADD COLUMN IF NOT EXISTS alpha TEXT",
-        "ALTER TABLE things ADD COLUMN IF NOT EXISTS beta INTEGER NOT NULL DEFAULT 0",
-        "ALTER TABLE things ADD COLUMN IF NOT EXISTS gamma TEXT",
+    const TEST_MIGRATIONS: &[Migration] = &[
+        Migration {
+            sql: "ALTER TABLE things ADD COLUMN alpha TEXT",
+            adds_column: Some(("things", "alpha")),
+        },
+        Migration {
+            sql: "ALTER TABLE things ADD COLUMN beta INTEGER NOT NULL DEFAULT 0",
+            adds_column: Some(("things", "beta")),
+        },
+        Migration {
+            sql: "ALTER TABLE things ADD COLUMN gamma TEXT",
+            adds_column: Some(("things", "gamma")),
+        },
     ];
 
     #[test]
     fn fresh_install_runs_all_migrations() {
         let conn = open_tmp();
         conn.execute_batch(TEST_SCHEMA).unwrap();
-        apply_versioned_migrations(&conn, "things", ("things", "gamma"), TEST_MIGRATIONS).unwrap();
+        apply_versioned_migrations(&conn, "things", TEST_MIGRATIONS).unwrap();
         // All three columns should exist.
         for col in ["alpha", "beta", "gamma"] {
             let exists: bool = conn
@@ -190,7 +222,7 @@ mod tests {
              alpha TEXT, beta INTEGER NOT NULL DEFAULT 0, gamma TEXT)",
         )
         .unwrap();
-        apply_versioned_migrations(&conn, "things", ("things", "gamma"), TEST_MIGRATIONS).unwrap();
+        apply_versioned_migrations(&conn, "things", TEST_MIGRATIONS).unwrap();
         // Version should be set to migrations.len() (3), not 0.
         let v: i64 = conn
             .query_row(
@@ -212,7 +244,7 @@ mod tests {
              INSERT INTO schema_versions (store, version) VALUES ('things', 1)",
         )
         .unwrap();
-        apply_versioned_migrations(&conn, "things", ("things", "gamma"), TEST_MIGRATIONS).unwrap();
+        apply_versioned_migrations(&conn, "things", TEST_MIGRATIONS).unwrap();
         // Only beta and gamma should have been added (alpha already existed).
         for col in ["alpha", "beta", "gamma"] {
             let exists: bool = conn
@@ -241,9 +273,9 @@ mod tests {
     fn already_at_latest_is_noop() {
         let conn = open_tmp();
         conn.execute_batch(TEST_SCHEMA).unwrap();
-        apply_versioned_migrations(&conn, "things", ("things", "gamma"), TEST_MIGRATIONS).unwrap();
+        apply_versioned_migrations(&conn, "things", TEST_MIGRATIONS).unwrap();
         // Second call is a no-op.
-        apply_versioned_migrations(&conn, "things", ("things", "gamma"), TEST_MIGRATIONS).unwrap();
+        apply_versioned_migrations(&conn, "things", TEST_MIGRATIONS).unwrap();
         let v: i64 = conn
             .query_row(
                 "SELECT version FROM schema_versions WHERE store = 'things'",
@@ -265,15 +297,19 @@ mod tests {
         apply_versioned_migrations(
             &conn,
             "store_a",
-            ("a", "col1"),
-            &["ALTER TABLE a ADD COLUMN col1 TEXT"],
+            &[Migration {
+                sql: "ALTER TABLE a ADD COLUMN col1 TEXT",
+                adds_column: Some(("a", "col1")),
+            }],
         )
         .unwrap();
         apply_versioned_migrations(
             &conn,
             "store_b",
-            ("b", "colx"),
-            &["ALTER TABLE b ADD COLUMN colx INTEGER NOT NULL DEFAULT 0"],
+            &[Migration {
+                sql: "ALTER TABLE b ADD COLUMN colx INTEGER NOT NULL DEFAULT 0",
+                adds_column: Some(("b", "colx")),
+            }],
         )
         .unwrap();
         // Each store tracks independently.
@@ -296,23 +332,20 @@ mod tests {
     }
 
     /// A partially-migrated legacy database: some columns from the old
-    /// try-and-ignore path exist, but the legacy marker column (used to
-    /// fast-forward fresh schema_versions entries to `len`) is absent.
-    /// With idempotent `ADD COLUMN IF NOT EXISTS` migrations this must
-    /// succeed — the already-present columns are skipped and only the
-    /// missing ones are added.
+    /// try-and-ignore path exist, with no `schema_versions` row yet. The
+    /// per-migration `adds_column` guard must skip the already-present
+    /// columns and add only the missing ones — no "duplicate column name".
     #[test]
     fn partial_legacy_completes_without_error() {
         let conn = open_tmp();
         // Simulate a DB that has `alpha` (v1) but not `beta` (v2) or
-        // `gamma` (v3). No schema_versions row exists yet, and `gamma`
-        // (the legacy marker) is absent → the function cannot fast-forward
-        // and must start from v0 with idempotent migrations.
+        // `gamma` (v3). No schema_versions row exists → start from v0 and
+        // guard-skip alpha while adding beta + gamma.
         conn.execute_batch(
             "CREATE TABLE IF NOT EXISTS things (id INTEGER PRIMARY KEY, alpha TEXT)",
         )
         .unwrap();
-        apply_versioned_migrations(&conn, "things", ("things", "gamma"), TEST_MIGRATIONS).unwrap();
+        apply_versioned_migrations(&conn, "things", TEST_MIGRATIONS).unwrap();
         // All three columns must now exist.
         for col in ["alpha", "beta", "gamma"] {
             let exists: bool = conn
@@ -322,7 +355,10 @@ mod tests {
                 .unwrap()
                 .exists([])
                 .unwrap();
-            assert!(exists, "column {col} missing after partial-legacy migration");
+            assert!(
+                exists,
+                "column {col} missing after partial-legacy migration"
+            );
         }
         // Version must be at the latest.
         let v: i64 = conn
@@ -332,6 +368,9 @@ mod tests {
                 |r| r.get(0),
             )
             .unwrap();
-        assert_eq!(v, 3, "must reach latest version after completing partial legacy");
+        assert_eq!(
+            v, 3,
+            "must reach latest version after completing partial legacy"
+        );
     }
 }
diff --git a/hive-c0re/src/stores/operator_questions.rs b/hive-c0re/src/stores/operator_questions.rs
index 4f352020..64f7b350 100644
--- a/hive-c0re/src/stores/operator_questions.rs
+++ b/hive-c0re/src/stores/operator_questions.rs
@@ -19,6 +19,8 @@ use hive_sh4re::wire_time::now_unix;
 use rusqlite::{Connection, OptionalExtension, params};
 use serde::Serialize;
 
+use crate::db::Migration;
+
 const SCHEMA: &str = r"
 CREATE TABLE IF NOT EXISTS operator_questions (
     id           INTEGER PRIMARY KEY AUTOINCREMENT,
@@ -37,16 +39,25 @@ CREATE INDEX IF NOT EXISTS idx_operator_questions_pending
 /// `"operator_questions"`). Legacy databases are detected via the `target`
 /// column — the last column added before versioning — and fast-forwarded
 /// past all known migrations.
-const MIGRATIONS: &[&str] = &[
+const MIGRATIONS: &[Migration] = &[
     // v1: `multi` — checkbox-style multi-option questions.
-    "ALTER TABLE operator_questions ADD COLUMN IF NOT EXISTS \
+    Migration {
+        sql: "ALTER TABLE operator_questions ADD COLUMN \
      multi INTEGER NOT NULL DEFAULT 0",
+        adds_column: Some(("operator_questions", "multi")),
+    },
     // v2: `deadline_at` — optional TTL after which the watchdog auto-resolves.
-    "ALTER TABLE operator_questions ADD COLUMN IF NOT EXISTS deadline_at INTEGER",
+    Migration {
+        sql: "ALTER TABLE operator_questions ADD COLUMN deadline_at INTEGER",
+        adds_column: Some(("operator_questions", "deadline_at")),
+    },
     // v3: `target` — recipient of the question. NULL = operator (back-compat
     // default); non-null = peer-to-peer question. Dashboard's `pending()`
     // filters on `target IS NULL` so peer questions never leak to the operator.
-    "ALTER TABLE operator_questions ADD COLUMN IF NOT EXISTS target TEXT",
+    Migration {
+        sql: "ALTER TABLE operator_questions ADD COLUMN target TEXT",
+        adds_column: Some(("operator_questions", "target")),
+    },
 ];
 
 #[derive(Debug, Clone, Serialize)]
@@ -80,12 +91,7 @@ impl OperatorQuestions {
         let conn = crate::db::open(path, "operator_questions")?;
         conn.execute_batch(SCHEMA)
             .context("apply operator_questions schema")?;
-        crate::db::apply_versioned_migrations(
-            &conn,
-            "operator_questions",
-            ("operator_questions", "target"),
-            MIGRATIONS,
-        )?;
+        crate::db::apply_versioned_migrations(&conn, "operator_questions", MIGRATIONS)?;
         Ok(Self {
             conn: Mutex::new(conn),
         })
diff --git a/hive-c0re/src/stores/scheduled_prompts.rs b/hive-c0re/src/stores/scheduled_prompts.rs
index 4ffa132e..f0877a5c 100644
--- a/hive-c0re/src/stores/scheduled_prompts.rs
+++ b/hive-c0re/src/stores/scheduled_prompts.rs
@@ -20,6 +20,8 @@ use hive_sh4re::wire_time::now_unix;
 use rusqlite::{Connection, OptionalExtension, params};
 use serde::{Deserialize, Serialize};
 
+use crate::db::Migration;
+
 /// Typed error returned by [`ScheduledPrompts::pause`] and
 /// [`ScheduledPrompts::resume`] when the target row does not exist or
 /// is already cancelled. Handlers downcast on this type to emit 404
@@ -186,17 +188,18 @@ impl ScheduledPrompts {
             .context("enable foreign keys")?;
         conn.execute_batch(SCHEMA)
             .context("apply scheduled_prompts schema")?;
-        // Versioned migration: add paused_at_unix. Legacy databases are
-        // detected via the presence of this column (it was the only
-        // migration before versioning was introduced).
+        // Versioned migration: add paused_at_unix. The per-migration
+        // `adds_column` guard skips the ALTER on a legacy DB that already
+        // has the column (it was the only migration before versioning).
         crate::db::apply_versioned_migrations(
             &conn,
             "scheduled_prompts",
-            ("scheduled_prompts", "paused_at_unix"),
             &[
                 // v1: add paused_at_unix for per-schedule pause support.
-                "ALTER TABLE scheduled_prompts \
-                 ADD COLUMN IF NOT EXISTS paused_at_unix INTEGER",
+                Migration {
+                    sql: "ALTER TABLE scheduled_prompts ADD COLUMN paused_at_unix INTEGER",
+                    adds_column: Some(("scheduled_prompts", "paused_at_unix")),
+                },
             ],
         )?;
         // Migration: recreate the due-rows index to also exclude paused