//! Schema migrations for cqs index database
//!
//! When the schema version changes, migrations allow upgrading existing indexes
//! without requiring a full rebuild (`cqs index --force`).
//!
//! ## Adding a new migration
//!
//! 1. Increment `CURRENT_SCHEMA_VERSION` in `helpers/mod.rs`
//! 2. Add a new migration function:
//!    `async fn migrate_vN_to_vM(conn: &mut sqlx::SqliteConnection) -> Result<(), StoreError>`
//! 3. Append a row to the `MIGRATIONS` table:
//!    `(N, M, |c| Box::pin(migrate_vN_to_vM(c)))`
//! 4. Update `schema.sql` with the new schema
//!
//! ## Migration guidelines
//!
//! - Most changes are additive (new columns, new tables) - these preserve data
//! - For new columns with NOT NULL, use DEFAULT or populate from existing data
//! - Test migrations with real indexes before release
//! - Keep migrations idempotent where possible (use IF NOT EXISTS)

use std::path::Path;

use sqlx::SqlitePool;

use super::backup;
use super::helpers::StoreError;

// Used by tests and future migrations
#[allow(unused_imports)]
use super::helpers::CURRENT_SCHEMA_VERSION;

/// Run all migrations from stored version to current version.
///
/// ## Backup and recovery
///
/// Before any DDL runs, a filesystem snapshot of `db_path` (and its
/// `-wal`/`-shm` sidecars) is taken to a sibling `.bak-v{from}-v{to}-{ts}.db`
/// file via `crate::fs::atomic_replace`. This covers two failure modes that
/// SQLite's transactional rollback does not:
///
/// 1. A commit-time I/O failure mid-WAL-write (disk full, fs quota, network
///    FS disconnect, user pulling USB). The in-memory pool state can think
///    the rollback completed while the on-disk file sees partial pages.
/// 2. A bug inside a migration function that writes logically-inconsistent
///    state — the transaction commits cleanly but the data is wrong.
///
/// On any migration error, the DB is restored from the backup atomically so
/// the caller sees either pre-migrate or post-migrate state, never a
/// partial write. On success, the newest two backups (including this one)
/// are retained and older ones are pruned.
///
/// If the backup step itself fails (e.g. parent dir is read-only, disk full),
/// the migration aborts by default with `StoreError::Io` — destructive
/// migrations (v18→v19 drops the old `sparse_vectors` table) without a
/// recovery snapshot are a data-loss hazard on a subsequent commit failure.
/// Setting `CQS_MIGRATE_REQUIRE_BACKUP=0` downgrades that to a `warn!` and
/// proceeds without a snapshot for users who accept the risk (tight-quota
/// filesystems, CI rebuilding from source).
///
/// ## Concurrent-migrate safety
///
/// Re-reads `schema_version` inside the migration transaction before executing
/// DDL. Two concurrent processes both reading version=17 from the pool, then
/// both running `ALTER TABLE ADD COLUMN`, would crash the second with
/// "duplicate column name" on a perfectly healthy DB. The double-check under
/// the transaction's implicit exclusive lock prevents this: the second
/// process sees the version has already advanced and short-circuits.
///
/// ## Pool ownership
///
/// `migrate` takes ownership of `pool` by value because the failure path must
/// `.close().await` the pool before `restore_from_backup` runs `atomic_replace`
/// over `db_path`. SQLite's in-process pool holds file descriptors against the
/// old inode; if those descriptors stay open across the file replace, queries
/// through them see the unlinked-old inode while readers from new processes
/// see the restored DB. The WAL/SHM sidecars copied alongside the main DB land
/// on the new inode, but the pool's mmap'd sidecars belong to the old —
/// silent two-state divergence.
///
/// Returns:
/// - `Ok(pool)` on success: the pool is the same one passed in, still usable.
/// - `Err(_)` on failure: the pool has been **consumed and closed** as part of
///   the restore protocol (when a backup was taken) or is dropped silently
///   (when no backup was available). The caller must reopen a fresh pool
///   against `db_path` to continue. The DB on disk is in its pre-migrate state
///   on the with-backup path, or in whatever state the rolled-back transaction
///   left it on the no-backup path (typically pre-migrate because all DDL ran
///   inside a single `pool.begin()`).
pub async fn migrate(
    pool: SqlitePool,
    db_path: &Path,
    from: i32,
    to: i32,
) -> Result<SqlitePool, StoreError> {
    let _span = tracing::info_span!("migrate", from, to).entered();

    if from == to {
        // Fast path: no work to do. Do NOT take a backup — this path runs
        // on every `cqs` command when the DB is already at the current
        // version, and a disk write here would be unacceptable overhead.
        return Ok(pool);
    }
    if from > to {
        return Err(StoreError::SchemaNewerThanCq(from));
    }

    tracing::info!(
        from_version = from,
        to_version = to,
        "Starting schema migration"
    );

    // Snapshot the DB before any DDL runs. On failure the restore path uses
    // `atomic_replace` to put the DB back in its pre-migrate state.
    //
    // Borrowing &pool here is fine: backup_before_migrate runs a
    // `wal_checkpoint(FULL)` and a file copy; we still own the pool when it
    // returns, and the failure-path close-and-restore happens below.
    let backup_path = backup::backup_before_migrate(&pool, db_path, from, to).await?;

    match run_migration_tx(&pool, from, to).await {
        Ok(()) => {
            tracing::info!(new_version = to, "Schema migration complete");
            // Best-effort prune of older backups; failure here is not a
            // migration failure — the user's DB is at the correct version.
            if let Err(e) = backup::prune_old_backups(db_path) {
                tracing::warn!(error = %e, "Failed to prune old migration backups");
            }
            Ok(pool)
        }
        Err(e) => {
            // Close the pool BEFORE atomic_replace overwrites the DB
            // file. Otherwise pool descriptors keep mmap'ing the unlinked
            // old inode while subsequent opens see the restored backup —
            // silent two-state divergence. Drain WAL first so the on-disk
            // bytes after close reflect the post-DDL state we're about to
            // discard, and the restore overwrites a quiesced file.
            if let Some(ref bak) = backup_path {
                if let Err(checkpoint_err) = sqlx::query("PRAGMA wal_checkpoint(TRUNCATE)")
                    .execute(&pool)
                    .await
                {
                    tracing::warn!(
                        error = %checkpoint_err,
                        "wal_checkpoint(TRUNCATE) before restore failed (non-fatal)"
                    );
                }
                pool.close().await;
                // The pool is closed; descriptors against the old inode are
                // released. Now atomic_replace can safely swap the DB file.

                match backup::restore_from_backup(db_path, bak) {
                    Ok(()) => {
                        tracing::warn!(
                            error = %e,
                            backup = %bak.display(),
                            "Migration failed; pool closed and DB restored from backup"
                        );
                    }
                    Err(restore_err) => {
                        // We can't put the DB back. Surface both errors in
                        // the log — the user needs to manually `cqs index
                        // --force` or copy `bak` into place.
                        tracing::error!(
                            migration_error = %e,
                            restore_error = %restore_err,
                            backup = %bak.display(),
                            db = %db_path.display(),
                            "Migration failed AND restore failed. \
                             Manually copy the backup into place or run \
                             'cqs index --force'."
                        );
                    }
                }
                Err(e)
            } else {
                // No backup was taken (env opt-out): the migration's DDL ran
                // inside a transaction that already rolled back, so the DB
                // file is still in its pre-migrate state. The pool was never
                // closed, but we drop it here to keep the API uniform —
                // callers receiving Err must reopen regardless of which
                // failure path fired.
                tracing::warn!(
                    error = %e,
                    db = %db_path.display(),
                    "Migration failed and no backup was available for restore. \
                     Run 'cqs index --force' to rebuild from source."
                );
                drop(pool);
                Err(e)
            }
        }
    }
}

/// Read the stored `schema_version` and migrate to the current version if
/// needed. Designed to be called from `Store::open` *before* the `Store`
/// struct is constructed so the pool can be handed off to `migrate()` by
/// value — a hard requirement of the close-and-restore protocol.
///
/// Returns the (possibly-the-same, possibly-migrated) pool on success.
/// On migration failure the pool is consumed (see [`migrate`]); the caller
/// must reopen if they want to continue.
///
/// `current_version` is normally [`super::helpers::CURRENT_SCHEMA_VERSION`],
/// passed in as a parameter so tests can target intermediate versions
/// without flipping a process-global constant.
pub async fn check_and_migrate_schema(
    pool: SqlitePool,
    db_path: &Path,
    current_version: i32,
    read_only: bool,
) -> Result<SqlitePool, StoreError> {
    let _span = tracing::info_span!("check_and_migrate_schema", read_only).entered();

    // Read the stored schema version. A "no such table" error means the
    // metadata table hasn't been created yet (fresh DB pre-init), which is
    // a legitimate post-open state — return the pool untouched.
    let row: Option<(String,)> =
        match sqlx::query_as("SELECT value FROM metadata WHERE key = 'schema_version'")
            .fetch_optional(&pool)
            .await
        {
            Ok(r) => r,
            Err(sqlx::Error::Database(e)) if e.message().contains("no such table") => {
                return Ok(pool);
            }
            Err(e) => return Err(e.into()),
        };

    let version: i32 = match row {
        Some((s,)) => s.parse().map_err(|e| {
            StoreError::Corruption(format!(
                "schema_version '{}' is not a valid integer: {}",
                s, e
            ))
        })?,
        // Missing key is OK — init() hasn't been called yet on a fresh DB.
        // After init(), schema_version is guaranteed present.
        None => 0,
    };

    if version > current_version {
        return Err(StoreError::SchemaNewerThanCq(version));
    }
    if version <= 0 || version >= current_version {
        // Either fresh-DB sentinel (0) or already current — no migration.
        return Ok(pool);
    }

    // Read-only handles can't migrate (every migration step issues writes).
    // Surface `SchemaMismatch` so the operator runs `cqs index` (or restarts
    // the watch daemon, which opens read-write and migrates structurally)
    // rather than triggering a backup → migrate → write-fail → restore loop
    // that leaves the DB at its old version, scatters `index.bak-*` snapshots,
    // and returns a confusing "attempt to write a readonly database" error.
    if read_only {
        tracing::info!(
            path = %db_path.display(),
            found = version,
            expected = current_version,
            "Read-only handle with stale schema; skipping migration (run `cqs index` to migrate)"
        );
        return Err(StoreError::SchemaMismatch {
            db_path: db_path.display().to_string(),
            found: version,
            expected: current_version,
        });
    }

    // Migration needed. Hand the pool off by value so migrate() can close it
    // around the file replace. On failure the pool is consumed; surface
    // SchemaMismatch for unsupported migrations so the CLI gets a clearer
    // error than the raw MigrationNotSupported (which encodes from/to as
    // anonymous integers).
    match migrate(pool, db_path, version, current_version).await {
        Ok(p) => {
            tracing::info!(
                path = %db_path.display(),
                from = version,
                to = current_version,
                "Schema migrated successfully"
            );
            Ok(p)
        }
        Err(StoreError::MigrationNotSupported { from, to }) => Err(StoreError::SchemaMismatch {
            db_path: db_path.display().to_string(),
            found: from,
            expected: to,
        }),
        Err(e) => Err(e),
    }
}

/// Run the migration transaction: re-check version under the write lock,
/// dispatch each version step, stamp the new `schema_version`, commit.
///
/// Split out of `migrate()` so the caller can always invoke the
/// backup-and-restore pipeline regardless of how the transaction fails.
async fn run_migration_tx(pool: &SqlitePool, from: i32, to: i32) -> Result<(), StoreError> {
    let mut tx = pool.begin().await?;

    // Re-read version under the write lock. A concurrent process may have
    // already migrated between our caller's pool-level read and our
    // transaction start. If the version is already at or past `to`, bail.
    let current_in_tx: Option<(String,)> =
        sqlx::query_as("SELECT value FROM metadata WHERE key = 'schema_version'")
            .fetch_optional(&mut *tx)
            .await?;
    let actual_from: i32 = current_in_tx
        .and_then(|(s,)| s.parse().ok())
        .unwrap_or(from);
    if actual_from >= to {
        tracing::info!(
            actual_from,
            to,
            "Schema already migrated by another process, skipping"
        );
        tx.rollback().await?;
        return Ok(());
    }

    for version in actual_from..to {
        tracing::info!(from = version, to = version + 1, "Running migration step");
        run_migration(&mut tx, version, version + 1).await?;

        // Test-only hook: when the thread-local `TEST_FAIL_AFTER_VERSION`
        // is set to N by a test, return an error without committing so
        // the test can exercise the backup-restore path. Per-thread (not
        // a process-global atomic or env var) so tests running in parallel
        // don't inject failures into each other.
        //
        // Gated on `cfg(test)` so it cannot be triggered in a release binary.
        #[cfg(test)]
        {
            let target = tests::TEST_FAIL_AFTER_VERSION.with(|c| c.get());
            if target != 0 && target == version + 1 {
                return Err(StoreError::Runtime(format!(
                    "test hook: injected failure after migration step v{} -> v{}",
                    version,
                    version + 1
                )));
            }
        }
    }
    // UPSERT (not UPDATE) so a DB without an existing `schema_version`
    // metadata row gets one stamped on first migration. A plain UPDATE
    // affects zero rows here, leaving the version unstamped and causing the
    // next open to re-run the same DDL ("duplicate column name" / "table
    // already exists" errors). Mirrors the `splade_generation` UPSERT
    // pattern in the per-version migrations.
    sqlx::query(
        "INSERT INTO metadata (key, value) VALUES ('schema_version', ?1)
         ON CONFLICT(key) DO UPDATE SET value = excluded.value",
    )
    .bind(to.to_string())
    .execute(&mut *tx)
    .await?;
    tx.commit().await?;

    Ok(())
}

/// Registered migration step. Each row pairs a `(from, to)` pair with a
/// function that builds a boxed future running the step. The
/// `Pin<Box<dyn Future>>` shape lets us hold a slice of `fn` pointers (no
/// closures, no trait objects), so adding a step is one row append rather
/// than editing a hand-coded `match` ladder.
// `+ Send` is intentionally absent — every per-version migration enters a
// `tracing::info_span!(...).entered()` whose `EnteredSpan` guard is `!Send`,
// so the resulting future isn't `Send` either. `run_migration` is awaited
// from the same task that holds the SQLite connection, so a non-`Send`
// future is fine. If a future migration moves the connection across
// `tokio::spawn`, that step would need its own restructure regardless.
type MigrationFn = for<'c> fn(
    &'c mut sqlx::SqliteConnection,
) -> std::pin::Pin<
    Box<dyn std::future::Future<Output = Result<(), StoreError>> + 'c>,
>;

const MIGRATIONS: &[(i32, i32, MigrationFn)] = &[
    (10, 11, |c| Box::pin(migrate_v10_to_v11(c))),
    (11, 12, |c| Box::pin(migrate_v11_to_v12(c))),
    (12, 13, |c| Box::pin(migrate_v12_to_v13(c))),
    (13, 14, |c| Box::pin(migrate_v13_to_v14(c))),
    (14, 15, |c| Box::pin(migrate_v14_to_v15(c))),
    (15, 16, |c| Box::pin(migrate_v15_to_v16(c))),
    (16, 17, |c| Box::pin(migrate_v16_to_v17(c))),
    (17, 18, |c| Box::pin(migrate_v17_to_v18(c))),
    (18, 19, |c| Box::pin(migrate_v18_to_v19(c))),
    (19, 20, |c| Box::pin(migrate_v19_to_v20(c))),
    (20, 21, |c| Box::pin(migrate_v20_to_v21(c))),
    (21, 22, |c| Box::pin(migrate_v21_to_v22(c))),
    (22, 23, |c| Box::pin(migrate_v22_to_v23(c))),
    (23, 24, |c| Box::pin(migrate_v23_to_v24(c))),
    (24, 25, |c| Box::pin(migrate_v24_to_v25(c))),
    (25, 26, |c| Box::pin(migrate_v25_to_v26(c))),
    (26, 27, |c| Box::pin(migrate_v26_to_v27(c))),
    (27, 28, |c| Box::pin(migrate_v27_to_v28(c))),
    (28, 29, |c| Box::pin(migrate_v28_to_v29(c))),
    (29, 30, |c| Box::pin(migrate_v29_to_v30(c))),
    (30, 31, |c| Box::pin(migrate_v30_to_v31(c))),
    (31, 32, |c| Box::pin(migrate_v31_to_v32(c))),
];

/// Run a single migration step
async fn run_migration(
    conn: &mut sqlx::SqliteConnection,
    from: i32,
    to: i32,
) -> Result<(), StoreError> {
    match MIGRATIONS.iter().find(|(f, t, _)| *f == from && *t == to) {
        Some((_, _, run)) => run(conn).await,
        None => Err(StoreError::MigrationNotSupported { from, to }),
    }
}

// ============================================================================
// Migration functions
// ============================================================================

/// Migrate from v10 to v11: add type_edges table
///
/// Adds type-level dependency tracking. Each edge records which chunk references
/// which type, with an edge_kind classification (Param, Return, Field, Impl, Bound, Alias).
/// Catch-all types (inside generics, etc.) use empty string '' for edge_kind instead of NULL
/// to simplify WHERE clause filtering.
///
/// The table will be empty after migration — run `cqs index --force` to populate.
async fn migrate_v10_to_v11(conn: &mut sqlx::SqliteConnection) -> Result<(), StoreError> {
    sqlx::query(
        "CREATE TABLE IF NOT EXISTS type_edges (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            source_chunk_id TEXT NOT NULL,
            target_type_name TEXT NOT NULL,
            edge_kind TEXT NOT NULL DEFAULT '',
            line_number INTEGER NOT NULL,
            FOREIGN KEY (source_chunk_id) REFERENCES chunks(id) ON DELETE CASCADE
        )",
    )
    .execute(&mut *conn)
    .await?;

    sqlx::query("CREATE INDEX IF NOT EXISTS idx_type_edges_source ON type_edges(source_chunk_id)")
        .execute(&mut *conn)
        .await?;
    sqlx::query("CREATE INDEX IF NOT EXISTS idx_type_edges_target ON type_edges(target_type_name)")
        .execute(&mut *conn)
        .await?;

    tracing::info!("Created type_edges table. Run 'cqs index --force' to populate type edges.");
    Ok(())
}

/// Migrate from v11 to v12: add parent_type_name column to chunks
///
/// Stores the enclosing class/struct/impl name for method chunks.
/// The column will be NULL after migration — run `cqs index --force` to populate.
async fn migrate_v11_to_v12(conn: &mut sqlx::SqliteConnection) -> Result<(), StoreError> {
    sqlx::query("ALTER TABLE chunks ADD COLUMN parent_type_name TEXT")
        .execute(&mut *conn)
        .await?;

    tracing::info!(
        "Added parent_type_name column. Run 'cqs index --force' to populate method→class links."
    );
    Ok(())
}

/// Migrate from v12 to v13: enrichment idempotency + HNSW dirty flag
///
/// - `enrichment_hash` column on chunks: blake3 hash of call context used during
///   enrichment. NULL means not yet enriched. Allows skipping already-enriched
///   chunks on re-index and detecting partial enrichment after crash.
/// - `hnsw_dirty` metadata key: set to "1" before SQLite chunk writes, cleared
///   to "0" after successful HNSW save. Detects crash between the two writes.
async fn migrate_v12_to_v13(conn: &mut sqlx::SqliteConnection) -> Result<(), StoreError> {
    sqlx::query("ALTER TABLE chunks ADD COLUMN enrichment_hash TEXT")
        .execute(&mut *conn)
        .await?;

    sqlx::query("INSERT OR IGNORE INTO metadata (key, value) VALUES ('hnsw_dirty', '0')")
        .execute(&mut *conn)
        .await?;

    tracing::info!(
        "Added enrichment_hash column and hnsw_dirty flag. \
         Run 'cqs index --force' to populate enrichment hashes."
    );
    Ok(())
}

/// Migrate from v13 to v14: LLM summaries cache table (SQ-6)
///
/// Stores one-sentence LLM-generated summaries keyed by content_hash.
/// Summaries survive chunk deletion and --force rebuilds.
async fn migrate_v13_to_v14(conn: &mut sqlx::SqliteConnection) -> Result<(), StoreError> {
    sqlx::query(
        "CREATE TABLE IF NOT EXISTS llm_summaries (
            content_hash TEXT PRIMARY KEY,
            summary TEXT NOT NULL,
            model TEXT NOT NULL,
            created_at TEXT NOT NULL
        )",
    )
    .execute(&mut *conn)
    .await?;

    tracing::info!("Created llm_summaries table for LLM-generated function summaries.");
    Ok(())
}

/// Migrate from v14 to v15: 768-dim embeddings (SQ-9)
///
/// Dropped the sentiment dimension — embeddings are now pure model-native output.
/// - Updates dimensions metadata from 769 to model dim (was 768 for E5-base-v2)
/// - Sets hnsw_dirty to trigger HNSW rebuild (old index has 769-dim vectors)
/// - Notes embedding column is left as-is (we write empty blobs now, old data is harmless)
async fn migrate_v14_to_v15(conn: &mut sqlx::SqliteConnection) -> Result<(), StoreError> {
    // Only update dimensions from 769→768 (the old sentiment-augmented size).
    // Databases already using a different model dim (e.g. 1024 for BGE-large) must
    // not be overwritten to 768.
    sqlx::query("UPDATE metadata SET value = '768' WHERE key = 'dimensions' AND value = '769'")
        .execute(&mut *conn)
        .await?;

    sqlx::query("UPDATE metadata SET value = '1' WHERE key = 'hnsw_dirty'")
        .execute(&mut *conn)
        .await?;

    tracing::info!(
        "Updated dimensions and marked HNSW dirty. \
         Run 'cqs index --force' to rebuild embeddings."
    );
    Ok(())
}

/// Migrate from v15 to v16: composite PK on llm_summaries (content_hash, purpose)
///
/// Recreates llm_summaries with a composite primary key so the same content_hash
/// can have multiple summaries for different purposes (e.g., 'summary', 'doc-comment').
/// Existing rows get purpose='summary' as the default.
///
/// Safety: CREATE TABLE, INSERT INTO ... SELECT, DROP TABLE, and ALTER TABLE RENAME
/// are all transactional in SQLite (they write to sqlite_master within the same
/// transaction). If any step fails, the entire migration rolls back and the original
/// llm_summaries table remains intact. The caller (`migrate`) wraps all steps in a
/// single BEGIN/COMMIT via `pool.begin()`.
async fn migrate_v15_to_v16(conn: &mut sqlx::SqliteConnection) -> Result<(), StoreError> {
    sqlx::query(
        "CREATE TABLE IF NOT EXISTS llm_summaries_v2 (
            content_hash TEXT NOT NULL,
            purpose TEXT NOT NULL DEFAULT 'summary',
            summary TEXT NOT NULL,
            model TEXT NOT NULL,
            created_at TEXT NOT NULL,
            PRIMARY KEY (content_hash, purpose)
        )",
    )
    .execute(&mut *conn)
    .await?;

    sqlx::query(
        "INSERT OR IGNORE INTO llm_summaries_v2 (content_hash, purpose, summary, model, created_at) \
         SELECT content_hash, 'summary', summary, model, created_at FROM llm_summaries",
    )
    .execute(&mut *conn)
    .await?;

    sqlx::query("DROP TABLE IF EXISTS llm_summaries")
        .execute(&mut *conn)
        .await?;

    sqlx::query("ALTER TABLE llm_summaries_v2 RENAME TO llm_summaries")
        .execute(&mut *conn)
        .await?;

    tracing::info!("Recreated llm_summaries with composite PK (content_hash, purpose).");
    Ok(())
}

/// Migrate from v16 to v17: sparse_vectors table + enrichment_version column
///
/// - `sparse_vectors`: stores SPLADE sparse vectors for hybrid search.
///   Each chunk gets a set of (token_id, weight) pairs from the learned sparse encoder.
/// - `enrichment_version`: idempotency marker. Tracks which enrichment pass
///   last processed each chunk, preventing double-application of call graph context.
async fn migrate_v16_to_v17(conn: &mut sqlx::SqliteConnection) -> Result<(), StoreError> {
    let _span = tracing::info_span!("migrate_v16_to_v17").entered();

    sqlx::query(
        "CREATE TABLE IF NOT EXISTS sparse_vectors (
            chunk_id TEXT NOT NULL,
            token_id INTEGER NOT NULL,
            weight REAL NOT NULL,
            PRIMARY KEY (chunk_id, token_id)
        )",
    )
    .execute(&mut *conn)
    .await?;

    sqlx::query("CREATE INDEX IF NOT EXISTS idx_sparse_token ON sparse_vectors(token_id)")
        .execute(&mut *conn)
        .await?;

    // enrichment idempotency marker
    sqlx::query("ALTER TABLE chunks ADD COLUMN enrichment_version INTEGER NOT NULL DEFAULT 0")
        .execute(&mut *conn)
        .await?;

    tracing::info!("Migrated to v17: sparse_vectors table + enrichment_version column");
    Ok(())
}

/// Migrate from v17 to v18: add embedding_base column to chunks
///
/// Dual embeddings for adaptive retrieval. Each chunk gets a second
/// embedding built from the raw NL description (without LLM summary or call-graph
/// enrichment). Conceptual/behavioral/negation queries route to the base index,
/// structural/multi-step queries keep the enriched index.
///
/// NULL is a valid state post-migration — chunks haven't been re-embedded yet.
/// The base HNSW index is only built once the column is populated; until then
/// the router silently falls back to the enriched index.
async fn migrate_v17_to_v18(conn: &mut sqlx::SqliteConnection) -> Result<(), StoreError> {
    let _span = tracing::info_span!("migrate_v17_to_v18").entered();

    sqlx::query("ALTER TABLE chunks ADD COLUMN embedding_base BLOB")
        .execute(&mut *conn)
        .await?;

    tracing::info!("Migrated to v18: embedding_base column (NULL until next index pass)");
    Ok(())
}

/// Migrate from v18 to v19: add FK(chunk_id) ON DELETE CASCADE to sparse_vectors
///
/// Makes the cascade invariant structural: any delete from `chunks` cascades
/// to `sparse_vectors` automatically, the same way `calls.source_chunk_id` and
/// `type_edges.source_chunk_id` cascade. Without the FK, chunk deletes leaked
/// orphan sparse rows that no query could reach — invalidation counters
/// attached to mutable state must be enforced at the schema layer, not
/// instrumented at specific call sites.
///
/// SQLite does not support `ALTER TABLE ADD FOREIGN KEY`, so we rebuild the
/// table in place. Orphan rows (sparse_vectors whose chunk_id no longer
/// exists in `chunks`) are dropped during the copy — they were leaked data
/// by definition. Row count before and after is logged for transparency.
///
/// After the table swap, the SPLADE generation counter is bumped
/// unconditionally so any persisted `splade.index.bin` from a pre-v19 schema
/// is invalidated on the next load (the on-disk file's embedded generation
/// won't match, forcing a clean rebuild from the new FK-protected table).
async fn migrate_v18_to_v19(conn: &mut sqlx::SqliteConnection) -> Result<(), StoreError> {
    let _span = tracing::info_span!("migrate_v18_to_v19").entered();

    // Count rows before the rebuild so the migration log makes any silent
    // orphan purge visible instead of invisible.
    let (before_rows,): (i64,) = sqlx::query_as("SELECT COUNT(*) FROM sparse_vectors")
        .fetch_one(&mut *conn)
        .await?;

    // Create the new table with the FK constraint. Same PRIMARY KEY shape as
    // v17 so the idx_sparse_token rebuild below is a drop-in replacement.
    sqlx::query(
        "CREATE TABLE sparse_vectors_v19 (
            chunk_id TEXT NOT NULL,
            token_id INTEGER NOT NULL,
            weight REAL NOT NULL,
            PRIMARY KEY (chunk_id, token_id),
            FOREIGN KEY (chunk_id) REFERENCES chunks(id) ON DELETE CASCADE
        )",
    )
    .execute(&mut *conn)
    .await?;

    // Copy only rows whose chunk_id exists in chunks — the INNER JOIN is the
    // orphan filter. Any row that doesn't match was already unreachable; the
    // migration is the right place to drop them.
    sqlx::query(
        "INSERT INTO sparse_vectors_v19 (chunk_id, token_id, weight)
         SELECT s.chunk_id, s.token_id, s.weight
         FROM sparse_vectors s
         INNER JOIN chunks c ON c.id = s.chunk_id",
    )
    .execute(&mut *conn)
    .await?;

    let (after_rows,): (i64,) = sqlx::query_as("SELECT COUNT(*) FROM sparse_vectors_v19")
        .fetch_one(&mut *conn)
        .await?;
    let dropped = before_rows - after_rows;
    // Escalate to `error!` when the INNER JOIN filter drops more than 10% of
    // rows. A small orphan set is expected; a large drop suggests the chunks
    // table was truncated out-of-band, or a bug produced widespread
    // chunk/sparse inconsistency. Not a hard fail — the rebuild is still an
    // improvement on the old unconstrained shape — but the user should know.
    // Integer math (not lossy f64 cast): require dropped > threshold AND
    // threshold > 0 so before_rows < 10 falls through to the warn arm rather
    // than tripping the error arm on a single dropped orphan.
    let threshold = before_rows / 10;
    if before_rows > 0 && threshold > 0 && dropped > threshold {
        tracing::error!(
            before = before_rows,
            after = after_rows,
            dropped_orphans = dropped,
            threshold,
            "v18→v19 migration dropped more than 10% of sparse_vectors rows \
             as orphans — this may indicate prior corruption or an out-of-band \
             chunks truncation. Review the log and consider rebuilding from \
             source via 'cqs index --force' if the drop looks wrong."
        );
    } else if dropped > 0 {
        tracing::warn!(
            before = before_rows,
            after = after_rows,
            dropped_orphans = dropped,
            "v18→v19 migration dropped orphan sparse_vectors rows (chunks no longer exist). \
             These were leaks from pre-v19 delete paths."
        );
    } else {
        tracing::info!(
            rows = before_rows,
            "v18→v19 sparse_vectors row count unchanged after FK filter"
        );
    }

    // Drop the old idx_sparse_token first (it's tied to the old table) so the
    // swap doesn't trip the UNIQUE-index-name constraint.
    sqlx::query("DROP INDEX IF EXISTS idx_sparse_token")
        .execute(&mut *conn)
        .await?;
    sqlx::query("DROP TABLE sparse_vectors")
        .execute(&mut *conn)
        .await?;
    sqlx::query("ALTER TABLE sparse_vectors_v19 RENAME TO sparse_vectors")
        .execute(&mut *conn)
        .await?;
    sqlx::query("CREATE INDEX idx_sparse_token ON sparse_vectors(token_id)")
        .execute(&mut *conn)
        .await?;

    // Bump splade_generation so any on-disk splade.index.bin from pre-v19
    // is invalidated — its header generation won't match the new value.
    // The UPSERT seeds the row if it wasn't present (older DBs without the
    // counter never had this metadata key).
    sqlx::query(
        "INSERT INTO metadata (key, value) VALUES ('splade_generation', '1')
         ON CONFLICT(key) DO UPDATE SET
             value = CAST((CAST(value AS INTEGER) + 1) AS TEXT)",
    )
    .execute(&mut *conn)
    .await?;

    tracing::info!(
        "Migrated to v19: sparse_vectors has FK(chunk_id) → chunks(id) ON DELETE CASCADE"
    );
    Ok(())
}

/// Migrate from v19 to v20: AFTER DELETE trigger on chunks bumps splade_generation
///
/// Guarantees that a chunk delete invalidates the SPLADE index. The v19 FK
/// CASCADE removes orphan sparse rows when `cqs watch` edits a file, but the
/// CASCADE alone does not bump `splade_generation`, so the on-disk
/// `splade.index.bin` would still match the unchanged counter and readers
/// would serve chunk_ids that no longer exist.
///
/// This trigger fires on every `DELETE FROM chunks` statement, once per
/// deleted row, and bumps the generation via a single metadata UPDATE.
/// For a watch cycle that touches 1-200 chunks, that's 1-200 metadata
/// updates — negligible. For `cqs index --force`, the new DB is fresh
/// and receives no DELETE statements at all, so the trigger cost is
/// zero on the bulk-reindex path. The only concern is `delete_by_origin`
/// during normal reindex when many chunks are displaced; even then the
/// write amplification is ~1-5s per 10k deletions, vs. the minutes the
/// actual rebuild takes.
///
/// The trigger is scoped to `chunks` deletions specifically. sparse_vectors
/// writes are still bumped explicitly by `bump_splade_generation_tx` (one
/// call per upsert transaction, not per row) because that's the only site
/// the trigger-on-sparse_vectors alternative would have caught and it
/// would have fired millions of times during a bulk upsert — row-level
/// triggers on the high-cardinality table were the wrong design.
async fn migrate_v19_to_v20(conn: &mut sqlx::SqliteConnection) -> Result<(), StoreError> {
    let _span = tracing::info_span!("migrate_v19_to_v20").entered();

    sqlx::query(
        "CREATE TRIGGER IF NOT EXISTS bump_splade_on_chunks_delete \
         AFTER DELETE ON chunks \
         BEGIN \
             INSERT INTO metadata (key, value) VALUES ('splade_generation', '1') \
             ON CONFLICT(key) DO UPDATE SET \
                 value = CAST((CAST(value AS INTEGER) + 1) AS TEXT); \
         END",
    )
    .execute(&mut *conn)
    .await?;

    // Bump generation immediately so any pre-v20 persisted splade.index.bin
    // (possibly already out of sync with sparse_vectors because v19 watch-mode
    // deletes never bumped the counter) gets invalidated on the next load.
    sqlx::query(
        "INSERT INTO metadata (key, value) VALUES ('splade_generation', '1')
         ON CONFLICT(key) DO UPDATE SET
             value = CAST((CAST(value AS INTEGER) + 1) AS TEXT)",
    )
    .execute(&mut *conn)
    .await?;

    tracing::info!(
        "Migrated to v20: AFTER DELETE trigger on chunks bumps splade_generation (DS-W2/OB-22 fix)"
    );
    Ok(())
}

/// Migrate from v20 to v21: add `parser_version` column to chunks
///
/// The watch path UPSERTs chunks with an
/// `ON CONFLICT(id) DO UPDATE ... WHERE chunks.content_hash != excluded.content_hash`
/// short-circuit, which is correct when the only thing that changes is the
/// source bytes. But `extract_doc_fallback_for_short_chunk` can change `doc`
/// for a chunk whose source bytes are byte-identical to a previously-indexed
/// version — without a parser stamp that change is silently discarded on
/// incremental update. A `parser_version` stamp lets the UPSERT invalidate
/// rows whose parser logic moved on, mirroring `splade_generation`.
///
/// Defaults to 0 for existing rows so the next `cqs index` (or watch reindex)
/// will write the live PARSER_VERSION value and refresh the affected fields.
async fn migrate_v20_to_v21(conn: &mut sqlx::SqliteConnection) -> Result<(), StoreError> {
    let _span = tracing::info_span!("migrate_v20_to_v21").entered();

    sqlx::query("ALTER TABLE chunks ADD COLUMN parser_version INTEGER NOT NULL DEFAULT 0")
        .execute(&mut *conn)
        .await?;

    tracing::info!(
        "Migrated to v21: parser_version column on chunks (P2 #29 — content-hash-stable doc refresh)"
    );
    Ok(())
}

/// Migrate from v21 to v22: add umap_x and umap_y REAL columns to chunks.
///
/// Both columns are nullable (REAL is nullable by default in SQLite). They
/// stay NULL until `cqs index --umap` runs the umap-learn projection
/// (`scripts/run_umap.py`) over the persisted chunk embeddings and writes
/// the 2D coordinates back. The /api/embed/2d endpoint in `cqs serve`
/// filters to `umap_x IS NOT NULL`, so the cluster view is dark until the
/// projection has been computed at least once but the rest of cqs is
/// unaffected.
async fn migrate_v21_to_v22(conn: &mut sqlx::SqliteConnection) -> Result<(), StoreError> {
    let _span = tracing::info_span!("migrate_v21_to_v22").entered();

    sqlx::query("ALTER TABLE chunks ADD COLUMN umap_x REAL")
        .execute(&mut *conn)
        .await?;
    sqlx::query("ALTER TABLE chunks ADD COLUMN umap_y REAL")
        .execute(&mut *conn)
        .await?;

    tracing::info!(
        "Migrated to v22: umap_x/umap_y columns on chunks (cqs serve cluster view, opt-in via `cqs index --umap`)"
    );
    Ok(())
}

/// v22 → v23: Add `source_size` (INTEGER) + `source_content_hash` (BLOB)
/// columns on `chunks` to power the reconcile fingerprint. Both nullable:
/// pre-v23 rows stay valid; first re-embed populates them. Reconcile uses
/// these as tiebreakers when mtimes are
/// identical (FAT32/NTFS/HFS+/SMB ≥1s mtime resolution) or when mtime
/// changed but content didn't (`git checkout`, formatter passes).
async fn migrate_v22_to_v23(conn: &mut sqlx::SqliteConnection) -> Result<(), StoreError> {
    let _span = tracing::info_span!("migrate_v22_to_v23").entered();

    sqlx::query("ALTER TABLE chunks ADD COLUMN source_size INTEGER")
        .execute(&mut *conn)
        .await?;
    sqlx::query("ALTER TABLE chunks ADD COLUMN source_content_hash BLOB")
        .execute(&mut *conn)
        .await?;

    tracing::info!(
        "Migrated to v23: source_size + source_content_hash columns on chunks (reconcile fingerprint, #1219)"
    );
    Ok(())
}

/// v23 → v24: Add `vendored` (INTEGER NOT NULL DEFAULT 0) column on
/// `chunks` to power the third-party-content trust-level downgrade.
/// Pre-migration rows default to `vendored = 0` (treated as user-code);
/// the next reindex flags chunks whose `origin` matches a configured
/// vendored-path prefix (`vendor/`, `node_modules/`, etc.).
/// Search/scout/onboard JSON output then emits
/// `trust_level: "vendored-code"` for those chunks instead of the bare
/// `user-code` claim.
async fn migrate_v23_to_v24(conn: &mut sqlx::SqliteConnection) -> Result<(), StoreError> {
    let _span = tracing::info_span!("migrate_v23_to_v24").entered();

    sqlx::query("ALTER TABLE chunks ADD COLUMN vendored INTEGER NOT NULL DEFAULT 0")
        .execute(&mut *conn)
        .await?;

    tracing::info!(
        "Migrated to v24: vendored column on chunks (trust-level downgrade for vendored content, #1221). \
         Existing rows default to vendored=0; reindex to flag chunks under configured vendored-path prefixes."
    );
    Ok(())
}

/// v24 → v25: Add `kind` (TEXT, nullable) column on `notes` for the
/// kind/tag taxonomy. Pre-v25 rows stay valid with `kind = NULL`; the
/// parser's `sentiment_to_prefix` mapping continues to drive
/// embedding-text prefixes when `kind = None`. New notes added via
/// `cqs notes add --kind <kind>` populate the column at insert time
/// and take precedence over the sentiment-based prefix.
///
/// An index on `notes(kind)` powers the `cqs notes list --kind` filter —
/// created at the same migration step to keep v25 a single self-contained
/// schema bump.
async fn migrate_v24_to_v25(conn: &mut sqlx::SqliteConnection) -> Result<(), StoreError> {
    let _span = tracing::info_span!("migrate_v24_to_v25").entered();

    sqlx::query("ALTER TABLE notes ADD COLUMN kind TEXT")
        .execute(&mut *conn)
        .await?;
    sqlx::query("CREATE INDEX IF NOT EXISTS idx_notes_kind ON notes(kind)")
        .execute(&mut *conn)
        .await?;

    tracing::info!(
        "Migrated to v25: kind column on notes (structured tag taxonomy, #1133). \
         Existing rows default to kind=NULL; new notes via `cqs notes add --kind <kind>` populate it."
    );
    Ok(())
}

/// v25 → v26: add composite `(source_type, origin)` index on `chunks` so
/// `list_stale_files` and `prune_missing_files` can satisfy their
/// `WHERE source_type = ? + DISTINCT origin` queries from a single index
/// pass. Without it SQLite probes `idx_chunks_source_type`, then row-visits
/// (or falls through to a full scan + sort). Mirrors the schema.sql change
/// so fresh DBs match.
async fn migrate_v25_to_v26(conn: &mut sqlx::SqliteConnection) -> Result<(), StoreError> {
    let _span = tracing::info_span!("migrate_v25_to_v26").entered();

    sqlx::query(
        "CREATE INDEX IF NOT EXISTS idx_chunks_source_type_origin \
         ON chunks(source_type, origin)",
    )
    .execute(&mut *conn)
    .await?;

    tracing::info!(
        "Migrated to v26: idx_chunks_source_type_origin composite index. \
         Speeds up reconcile / status --watch-fresh / GC at 50k+ chunk corpora. \
         PERF-V1.33-10 / #1371."
    );
    Ok(())
}

/// v26 → v27: add `chunks.needs_embedding` flag. Lets the pipeline skip the
/// wasted first-pass embed when `--llm-summaries` is set (which guarantees
/// `enrichment_pass` will overwrite every chunk's embedding anyway).
/// Chunks written without a real embedding get `needs_embedding = 1` and a
/// zero-vec sentinel in the `embedding` column; HNSW build + search paths
/// filter `WHERE needs_embedding = 0` so those chunks are invisible until
/// `enrichment_pass` lands their real vector and clears the flag.
async fn migrate_v26_to_v27(conn: &mut sqlx::SqliteConnection) -> Result<(), StoreError> {
    let _span = tracing::info_span!("migrate_v26_to_v27").entered();

    // Add the flag with a default of 0 so every existing row is treated
    // as embedded (which they are — pre-v27 there was no skip-first-pass
    // path). Fresh DBs use the same default via schema.sql.
    sqlx::query("ALTER TABLE chunks ADD COLUMN needs_embedding INTEGER NOT NULL DEFAULT 0")
        .execute(&mut *conn)
        .await?;

    // Partial index on `needs_embedding = 1` so the rare "find chunks that
    // still need embedding" lookup in `enrichment_pass` is O(needs)
    // instead of O(total_chunks). Most chunks are embedded most of the
    // time, so a partial index keeps the disk footprint negligible.
    sqlx::query(
        "CREATE INDEX IF NOT EXISTS idx_chunks_needs_embedding \
         ON chunks(needs_embedding) WHERE needs_embedding = 1",
    )
    .execute(&mut *conn)
    .await?;

    // Backfill `needs_embedding=1` for any pre-v27 row that has
    // `embedding_base IS NULL`. Combined with enrichment_pass
    // repopulating `embedding_base` from the row's `embedding` on first
    // hit, this trips the next `cqs index --force` or
    // `cqs index --llm-summaries` cycle into filling the missing base
    // bytes — restoring base-HNSW coverage for `cqs <q> --rerank` /
    // DenseBase routing on projects that pre-date the v17→v18 base-column
    // split.
    let backfilled = sqlx::query(
        "UPDATE chunks SET needs_embedding = 1 \
         WHERE embedding_base IS NULL AND needs_embedding = 0",
    )
    .execute(&mut *conn)
    .await?
    .rows_affected();

    if backfilled > 0 {
        tracing::info!(
            backfilled,
            "v27 migration: marked {backfilled} chunks with NULL embedding_base \
             as needs_embedding=1 — they'll re-embed on the next enrichment_pass \
             so base-HNSW coverage repopulates"
        );
    }

    tracing::info!(
        "Migrated to v27: chunks.needs_embedding flag + partial index. \
         Enables --llm-summaries first-pass-embed skip #1452."
    );
    Ok(())
}

/// v27 → v28: add `chunks.canonical_hash` column + index.
///
/// `canonical_hash` is the blake3 of a comment- and whitespace-normalized
/// form of the chunk content (see `parser::chunk::canonical_hash`). It keys
/// the embedding-reuse cache so a comment-only or formatting-only edit (which
/// changes `content_hash`, the store identity) can still reuse the prior
/// embedding instead of re-embedding the whole corpus.
///
/// Nullable: NULL means "unknown" for pre-v28 rows. Old rows simply miss the
/// canonical-keyed reuse until the next reindex writes the column — no
/// backfill is attempted (it would require re-parsing every chunk to recover
/// the tree-precise comment ranges). The reuse query skips
/// `WHERE canonical_hash IS NULL` rows, so a NULL is a clean cache miss, never
/// a wrong hit.
///
/// Note the upsert only rewrites a row when content/parser_version/vendored
/// changed, so a plain `cqs index` does not backfill untouched files; the
/// column fills in as files are edited, or all at once via `cqs index --force`.
async fn migrate_v27_to_v28(conn: &mut sqlx::SqliteConnection) -> Result<(), StoreError> {
    let _span = tracing::info_span!("migrate_v27_to_v28").entered();

    sqlx::query("ALTER TABLE chunks ADD COLUMN canonical_hash TEXT")
        .execute(&mut *conn)
        .await?;

    // Index powers the `WHERE canonical_hash IN (...)` embedding-reuse lookup
    // in `get_embeddings_by_canonical_hashes`. Mirrors `idx_chunks_content_hash`.
    sqlx::query("CREATE INDEX IF NOT EXISTS idx_chunks_canonical_hash ON chunks(canonical_hash)")
        .execute(&mut *conn)
        .await?;

    tracing::info!(
        "Migrated to v28: chunks.canonical_hash column + idx_chunks_canonical_hash. \
         Comment/whitespace-only edits now reuse embeddings instead of re-embedding. \
         Pre-v28 rows stay NULL (clean cache miss) until the next reindex."
    );
    Ok(())
}

/// v28 → v29: file_registry table #1774 + notes.sentiment CHECK constraint.
///
/// **file_registry.** A file that parses to zero chunks (comment-only source,
/// parser-emitted-empty) has no chunk row to carry the reconcile fingerprint,
/// so the staleness pre-filter saw no rows, treated it as un-indexed, and
/// re-parsed it on every `cqs index` run. The new `file_registry` table
/// (origin → source_mtime/size/content_hash, mirroring the chunk-row
/// fingerprint columns) persists the fingerprint independent of chunk rows;
/// the staleness readers UNION it so a zero-chunk origin surfaces a stored
/// fingerprint and skips the parse. Created empty: the next `cqs index` /
/// watch reindex populates it as zero-chunk files are seen. Backfilling from
/// chunk rows is unnecessary — files WITH chunks already carry the fingerprint
/// on their rows, and an absent registry row contributes nothing to the
/// readers' MAX()-over-UNION merge.
///
/// **notes.sentiment CHECK.** Sentiment is documented as DISCRETE (only -1,
/// -0.5, 0, 0.5, 1), but pre-v29 the schema accepted any REAL. The CHECK pins
/// the invariant at the schema layer (per the invalidation-must-be-enforced
/// principle: enforce in schema, not call-site validation alone). SQLite cannot
/// `ALTER TABLE ADD CONSTRAINT`, so the notes table is rebuilt in place
/// (CREATE new → INSERT … SELECT → DROP old → RENAME), all transactional within
/// the migration's single `pool.begin()`. Existing rows whose sentiment is
/// off-grid (e.g. a `-0.8` written by a pre-snap `parse_notes_str`) are
/// clamp-rewritten to the nearest discrete value during the copy so the rebuilt
/// table satisfies its own CHECK — `round(sentiment * 2) / 2` after a
/// `[-1, 1]` clamp lands exactly on the five legal values. A `warn!` reports
/// how many rows were rewritten.
async fn migrate_v28_to_v29(conn: &mut sqlx::SqliteConnection) -> Result<(), StoreError> {
    let _span = tracing::info_span!("migrate_v28_to_v29").entered();

    // --- Part 1: file_registry table ---
    sqlx::query(
        "CREATE TABLE IF NOT EXISTS file_registry (
            origin TEXT PRIMARY KEY,
            source_mtime INTEGER,
            source_size INTEGER,
            source_content_hash BLOB
        )",
    )
    .execute(&mut *conn)
    .await?;

    // --- Part 2: notes.sentiment CHECK constraint ---
    // Count off-grid rows up front so the migration log makes the clamp
    // visible instead of silent. The CHECK below uses the same five-value
    // IN-list the rebuilt table declares.
    let (off_grid,): (i64,) = sqlx::query_as(
        "SELECT COUNT(*) FROM notes \
         WHERE sentiment NOT IN (-1.0, -0.5, 0.0, 0.5, 1.0)",
    )
    .fetch_one(&mut *conn)
    .await?;

    // Rebuild the notes table with the CHECK. SQLite has no ADD CONSTRAINT, so
    // create a sibling with the constraint, copy rows (snapping off-grid
    // sentiment to the nearest discrete value via round(s*2)/2 over a [-1,1]
    // clamp — MAX/MIN clamp, then the round expression), drop the old, rename.
    // The `notes_fts` virtual table and its indexes are untouched (they key on
    // `id`, which is preserved verbatim).
    sqlx::query(
        "CREATE TABLE notes_v29 (
            id TEXT PRIMARY KEY,
            text TEXT NOT NULL,
            sentiment REAL NOT NULL CHECK (sentiment IN (-1.0, -0.5, 0.0, 0.5, 1.0)),
            mentions TEXT,
            embedding BLOB NOT NULL,
            source_file TEXT NOT NULL,
            file_mtime INTEGER NOT NULL,
            created_at TEXT NOT NULL,
            updated_at TEXT NOT NULL,
            kind TEXT
        )",
    )
    .execute(&mut *conn)
    .await?;

    sqlx::query(
        "INSERT INTO notes_v29 \
            (id, text, sentiment, mentions, embedding, source_file, file_mtime, created_at, updated_at, kind) \
         SELECT id, text, \
                round(MAX(-1.0, MIN(1.0, sentiment)) * 2.0) / 2.0, \
                mentions, embedding, source_file, file_mtime, created_at, updated_at, kind \
         FROM notes",
    )
    .execute(&mut *conn)
    .await?;

    sqlx::query("DROP TABLE notes").execute(&mut *conn).await?;
    sqlx::query("ALTER TABLE notes_v29 RENAME TO notes")
        .execute(&mut *conn)
        .await?;

    // Recreate the indexes the original `notes` table carried — they were tied
    // to the dropped table and don't survive the rename.
    sqlx::query("CREATE INDEX IF NOT EXISTS idx_notes_sentiment ON notes(sentiment)")
        .execute(&mut *conn)
        .await?;
    sqlx::query("CREATE INDEX IF NOT EXISTS idx_notes_kind ON notes(kind)")
        .execute(&mut *conn)
        .await?;

    if off_grid > 0 {
        tracing::warn!(
            rewritten = off_grid,
            "v28→v29 migration clamp-rewrote {off_grid} notes with off-grid sentiment \
             to the nearest discrete value (-1, -0.5, 0, 0.5, 1) to satisfy the new CHECK"
        );
    }

    tracing::info!(
        "Migrated to v29: file_registry table (#1774 — zero-chunk files persist a \
         reconcile fingerprint and stop re-parsing every run) + notes.sentiment CHECK \
         pinning the five discrete documented values."
    );
    Ok(())
}

/// v29 → v30: add `function_calls.edge_kind` column classifying call-graph
/// edge provenance.
///
/// The call-graph extractor emits several edge kinds with very different
/// evidentiary weight: syntactic `call_expression` (`call`, ground truth),
/// serde string-callback attributes (`serde_callback`), Rust macro token-tree
/// heuristics (`macro_heuristic`), bare fn-pointer arguments (`fn_pointer`),
/// and markdown cross-references (`doc_reference`). Before this column they
/// flatten into indistinguishable rows, so a consuming agent cannot weight a
/// direct call differently from a heuristic or a prose mention. Mirrors the
/// `type_edges.edge_kind` precedent.
///
/// Additive `ALTER TABLE ADD COLUMN` with a `'call'` default. Existing rows
/// take `'call'`, which is correct for the syntactic majority but WRONG for
/// the pre-existing serde/macro/fn-pointer/doc-reference edges — those
/// re-extract with the right kind on the next reindex (PARSER_VERSION is
/// bumped 5→6 alongside this migration). The staleness pre-filters treat
/// PARSER_VERSION drift as stale (see `origins_with_parser_drift`), so a plain
/// `cqs index` re-parses every drifted file and re-tags its edges — no
/// `--force` needed. Pre-reindex, those edges read as `'call'`; the surfaces
/// document the absence ⇒ `call` convention.
async fn migrate_v29_to_v30(conn: &mut sqlx::SqliteConnection) -> Result<(), StoreError> {
    let _span = tracing::info_span!("migrate_v29_to_v30").entered();

    sqlx::query("ALTER TABLE function_calls ADD COLUMN edge_kind TEXT NOT NULL DEFAULT 'call'")
        .execute(&mut *conn)
        .await?;

    tracing::info!(
        "Migrated to v30: function_calls.edge_kind column (default 'call'). \
         Pre-v30 serde/macro/fn-pointer/doc-reference edges read as 'call' \
         until the next reindex re-tags them; PARSER_VERSION drift (bumped \
         5→6) makes drifted files stale so a plain reindex re-parses them \
         (no --force needed)."
    );
    Ok(())
}

/// v30 → v31: add `file_registry.parse_failed_parser_version` (nullable INTEGER).
///
/// The drift loop-breaker. `origins_with_parser_drift` selects files whose
/// `chunks.parser_version` differs from the current parser, requeuing them on
/// every reconcile tick. A successful re-parse advances the stamp and heals the
/// drift; a file that CANNOT parse (IO/permission failure, tree-sitter
/// `ParseFailed`) never advances it, so at a `PARSER_VERSION` bump such a file
/// is re-queued forever. The existing `touch_mtime` loop-breaker heals only the
/// FINGERPRINT predicate (it touches `source_mtime` so disk == stored), not the
/// orthogonal drift predicate.
///
/// On parse failure both the watch reconcile path and the bulk `cqs index` path
/// record the parser version they failed at here. `origins_with_parser_drift`
/// excludes origins whose marker == the current parser version, so a file that
/// already failed at the current version is not re-queued by drift until its
/// content changes (any successful re-parse or content edit clears the marker).
/// Additive `ALTER TABLE ADD COLUMN`; pre-v31 rows read NULL (no recorded
/// failure ⇒ never suppressed), which is the correct default.
async fn migrate_v30_to_v31(conn: &mut sqlx::SqliteConnection) -> Result<(), StoreError> {
    let _span = tracing::info_span!("migrate_v30_to_v31").entered();

    sqlx::query("ALTER TABLE file_registry ADD COLUMN parse_failed_parser_version INTEGER")
        .execute(&mut *conn)
        .await?;

    tracing::info!(
        "Migrated to v31: file_registry.parse_failed_parser_version column \
         (nullable). Records the parser version a drifted-but-unparseable file \
         failed at so origins_with_parser_drift stops re-queuing it every \
         reconcile tick until its content changes."
    );
    Ok(())
}

/// v31 → v32: add the `candidate_edges` side-table for low-confidence call-graph
/// candidates that must NOT surface as callers.
///
/// A SEPARATE table — deliberately NOT a new `function_calls.edge_kind` value.
/// Every callers/callees/impact query SELECTs `function_calls` rows for a callee
/// with no kind exclusion, so a candidate stored there would read as a false
/// caller; and `CallEdgeKind::is_real_caller` is the complement of
/// `doc_reference`, so a new kind would default to "real caller" and put the
/// callee in the wrong dead tier. This table is callee-name-keyed and is never
/// joined by a graph query, so a candidate is invisible to the caller graph by
/// construction.
///
/// Created EMPTY on migrate (additive). The columns mirror what a candidate site
/// carries: `file` (source path), `callee_name` (the symbol the candidate points
/// at), `ref_line` (1-indexed line of the reference), and `candidate_kind` (the
/// candidate's provenance string). Indexes on `callee_name` and `file` match the
/// two access patterns a future consumer/cleanup path needs (callee-name lookup
/// and per-file delete-on-reindex). No PARSER_VERSION bump: nothing emits rows
/// into this table yet, so no re-extraction is required.
async fn migrate_v31_to_v32(conn: &mut sqlx::SqliteConnection) -> Result<(), StoreError> {
    let _span = tracing::info_span!("migrate_v31_to_v32").entered();

    sqlx::query(
        "CREATE TABLE IF NOT EXISTS candidate_edges (
            file TEXT NOT NULL,
            callee_name TEXT NOT NULL,
            ref_line INTEGER NOT NULL,
            candidate_kind TEXT NOT NULL
        )",
    )
    .execute(&mut *conn)
    .await?;

    sqlx::query(
        "CREATE INDEX IF NOT EXISTS idx_candidate_edges_callee ON candidate_edges(callee_name)",
    )
    .execute(&mut *conn)
    .await?;
    sqlx::query("CREATE INDEX IF NOT EXISTS idx_candidate_edges_file ON candidate_edges(file)")
        .execute(&mut *conn)
        .await?;

    tracing::info!(
        "Migrated to v32: candidate_edges side-table (callee-name-keyed; never \
         joined by callers/callees/impact). Empty on migrate; populated by a \
         later parser-emit pass. No PARSER_VERSION bump — nothing writes rows yet."
    );
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;
    use sqlx::sqlite::SqlitePoolOptions;

    std::thread_local! {
        /// Test-only failure-injection flag read by `run_migration_tx`.
        /// Tests that want to exercise the `migrate()` backup/restore path
        /// set this to `target_version`; after the migration step that
        /// bumps the version to that value completes, `run_migration_tx`
        /// returns an error without committing. `0` = disabled.
        ///
        /// Thread-local rather than global so parallel tests don't inject
        /// failures into each other — each `#[test]` runs on its own
        /// thread, and the hook only fires on threads that explicitly set
        /// this cell.
        pub(super) static TEST_FAIL_AFTER_VERSION: std::cell::Cell<i32> =
            const { std::cell::Cell::new(0) };
    }

    #[test]
    fn test_migration_not_supported_error() {
        // Verify unknown migrations produce clear errors
        let err = StoreError::MigrationNotSupported { from: 5, to: 6 };
        let msg = err.to_string();
        assert!(msg.contains("5"));
        assert!(msg.contains("6"));
    }

    #[test]
    fn test_current_schema_version_documented() {
        // Ensure the current version matches what we document
        assert_eq!(CURRENT_SCHEMA_VERSION, 32);
    }

    #[test]
    fn test_migrations_table_contiguous() {
        // Every step from the oldest migratable version (v10) up to
        // CURRENT_SCHEMA_VERSION must have a (v, v+1) row in MIGRATIONS.
        // A forgotten row fails here instead of at a user's upgrade.
        for v in 10..CURRENT_SCHEMA_VERSION {
            assert!(
                MIGRATIONS.iter().any(|(f, t, _)| *f == v && *t == v + 1),
                "MIGRATIONS is missing the (v{}, v{}) step — append a row \
                 when bumping CURRENT_SCHEMA_VERSION",
                v,
                v + 1
            );
        }
    }

    #[test]
    fn test_migrate_noop_same_version() {
        // Migration from N to N should be a no-op
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("test.db");

        rt.block_on(async {
            let pool = SqlitePoolOptions::new()
                .max_connections(1)
                .connect_with(
                    sqlx::sqlite::SqliteConnectOptions::new()
                        .filename(&db_path)
                        .create_if_missing(true),
                )
                .await
                .unwrap();

            let result = migrate(pool, &db_path, 15, 15).await;
            assert!(result.is_ok(), "same-version migration should be no-op");
        });
    }

    #[test]
    fn test_migrate_rejects_downgrade() {
        // from > to should error with SchemaNewerThanCq
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("test.db");

        rt.block_on(async {
            let pool = SqlitePoolOptions::new()
                .max_connections(1)
                .connect_with(
                    sqlx::sqlite::SqliteConnectOptions::new()
                        .filename(&db_path)
                        .create_if_missing(true),
                )
                .await
                .unwrap();

            let result = migrate(pool, &db_path, 15, 14).await;
            assert!(result.is_err(), "downgrade should fail");
            match result.unwrap_err() {
                StoreError::SchemaNewerThanCq(v) => assert_eq!(v, 15),
                other => panic!("Expected SchemaNewerThanCq, got: {:?}", other),
            }
        });
    }

    /// Read-only handles can't migrate. Stale-schema readonly opens should
    /// surface `SchemaMismatch` instead of attempting writes that fail with
    /// the SQLite "attempt to write a readonly database" error and leave
    /// stray `index.bak-*` files behind.
    #[test]
    fn test_check_and_migrate_schema_readonly_returns_mismatch() {
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("test.db");

        rt.block_on(async {
            // Set up a DB stamped at an old schema version (writable open).
            let pool = SqlitePoolOptions::new()
                .max_connections(1)
                .connect_with(
                    sqlx::sqlite::SqliteConnectOptions::new()
                        .filename(&db_path)
                        .create_if_missing(true),
                )
                .await
                .unwrap();
            sqlx::query("CREATE TABLE metadata (key TEXT PRIMARY KEY, value TEXT NOT NULL)")
                .execute(&pool)
                .await
                .unwrap();
            sqlx::query("INSERT INTO metadata (key, value) VALUES ('schema_version', '10')")
                .execute(&pool)
                .await
                .unwrap();
            pool.close().await;

            // Reopen read-only and call check_and_migrate_schema with
            // read_only=true; should bail with SchemaMismatch (not attempt
            // a migration that would fail with "attempt to write a readonly
            // database").
            let ro_pool = SqlitePoolOptions::new()
                .max_connections(1)
                .connect_with(
                    sqlx::sqlite::SqliteConnectOptions::new()
                        .filename(&db_path)
                        .read_only(true),
                )
                .await
                .unwrap();
            let result = check_and_migrate_schema(ro_pool, &db_path, 26, true).await;
            match result {
                Err(StoreError::SchemaMismatch {
                    found, expected, ..
                }) => {
                    assert_eq!(found, 10);
                    assert_eq!(expected, 26);
                }
                Ok(_) => panic!("expected SchemaMismatch on stale readonly handle"),
                Err(other) => {
                    panic!("expected SchemaMismatch on stale readonly handle, got: {other:?}")
                }
            }

            // No backup files should have been created (no migrate path
            // means no `index.bak-*` snapshots).
            let backup_count = std::fs::read_dir(dir.path())
                .unwrap()
                .filter(|e| {
                    e.as_ref()
                        .ok()
                        .and_then(|e| e.file_name().to_str().map(String::from))
                        .map(|n| n.starts_with("test.bak-"))
                        .unwrap_or(false)
                })
                .count();
            assert_eq!(
                backup_count, 0,
                "readonly stale-schema check must not create migration backups"
            );
        });
    }

    /// A non-numeric `schema_version` value must surface as
    /// `StoreError::Corruption`, not panic and not be silently treated as
    /// version 0. Guards the `s.parse::<i32>()` failure arm.
    #[test]
    fn test_check_and_migrate_schema_non_numeric_version_is_corruption() {
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("test.db");

        rt.block_on(async {
            let pool = SqlitePoolOptions::new()
                .max_connections(1)
                .connect_with(
                    sqlx::sqlite::SqliteConnectOptions::new()
                        .filename(&db_path)
                        .create_if_missing(true),
                )
                .await
                .unwrap();
            sqlx::query("CREATE TABLE metadata (key TEXT PRIMARY KEY, value TEXT NOT NULL)")
                .execute(&pool)
                .await
                .unwrap();
            // Garbage version string — must fail the i32 parse.
            sqlx::query("INSERT INTO metadata (key, value) VALUES ('schema_version', 'abc')")
                .execute(&pool)
                .await
                .unwrap();

            let result = check_and_migrate_schema(pool, &db_path, 26, false).await;
            match result {
                Err(StoreError::Corruption(msg)) => {
                    assert!(
                        msg.contains("abc"),
                        "corruption message should name the bad value: {msg}"
                    );
                }
                Ok(_) => panic!("non-numeric schema_version must not be treated as valid"),
                Err(other) => panic!("expected Corruption for non-numeric version, got: {other:?}"),
            }
        });
    }

    /// Read-only opens of a DB already at the current version must succeed
    /// (no migration, no error). Regression guard for the `read_only=true`
    /// branch added above so it doesn't reject every readonly open.
    #[test]
    fn test_check_and_migrate_schema_readonly_current_version_succeeds() {
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("test.db");

        rt.block_on(async {
            let pool = SqlitePoolOptions::new()
                .max_connections(1)
                .connect_with(
                    sqlx::sqlite::SqliteConnectOptions::new()
                        .filename(&db_path)
                        .create_if_missing(true),
                )
                .await
                .unwrap();
            sqlx::query("CREATE TABLE metadata (key TEXT PRIMARY KEY, value TEXT NOT NULL)")
                .execute(&pool)
                .await
                .unwrap();
            sqlx::query("INSERT INTO metadata (key, value) VALUES ('schema_version', '26')")
                .execute(&pool)
                .await
                .unwrap();
            pool.close().await;

            let ro_pool = SqlitePoolOptions::new()
                .max_connections(1)
                .connect_with(
                    sqlx::sqlite::SqliteConnectOptions::new()
                        .filename(&db_path)
                        .read_only(true),
                )
                .await
                .unwrap();
            let result = check_and_migrate_schema(ro_pool, &db_path, 26, true).await;
            assert!(
                result.is_ok(),
                "readonly open at current version must succeed, got: {:?}",
                result.err()
            );
        });
    }

    #[test]
    fn test_migrate_v10_to_v11_creates_type_edges() {
        // Full migration test: set up a v10 schema, run migration, verify type_edges exists
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("test.db");

        rt.block_on(async {
            let pool = SqlitePoolOptions::new()
                .max_connections(1)
                .connect_with(
                    sqlx::sqlite::SqliteConnectOptions::new()
                        .filename(&db_path)
                        .create_if_missing(true),
                )
                .await
                .unwrap();

            // Create the minimal schema that a v10 store would have
            sqlx::query(
                "CREATE TABLE IF NOT EXISTS metadata (
                    key TEXT PRIMARY KEY,
                    value TEXT NOT NULL
                )",
            )
            .execute(&pool)
            .await
            .unwrap();

            sqlx::query(
                "CREATE TABLE IF NOT EXISTS chunks (
                    id TEXT PRIMARY KEY,
                    origin TEXT NOT NULL,
                    language TEXT NOT NULL DEFAULT '',
                    chunk_type TEXT NOT NULL DEFAULT '',
                    name TEXT NOT NULL,
                    signature TEXT NOT NULL DEFAULT '',
                    content TEXT NOT NULL,
                    doc TEXT,
                    line_start INTEGER NOT NULL DEFAULT 0,
                    line_end INTEGER NOT NULL DEFAULT 0,
                    parent_id TEXT
                )",
            )
            .execute(&pool)
            .await
            .unwrap();

            // Set schema_version to 10
            sqlx::query("INSERT INTO metadata (key, value) VALUES ('schema_version', '10')")
                .execute(&pool)
                .await
                .unwrap();

            // Verify type_edges does NOT exist before migration
            let table_check: Option<(String,)> = sqlx::query_as(
                "SELECT name FROM sqlite_master WHERE type='table' AND name='type_edges'",
            )
            .fetch_optional(&pool)
            .await
            .unwrap();
            assert!(table_check.is_none(), "type_edges should not exist yet");

            // Run migration from v10 to v11. migrate consumes the pool by
            // value so the failure path can close it before the file replace;
            // we rebind to the returned pool to keep using it.
            let pool = migrate(pool, &db_path, 10, 11).await.unwrap();

            // Verify type_edges now exists
            let table_check: Option<(String,)> = sqlx::query_as(
                "SELECT name FROM sqlite_master WHERE type='table' AND name='type_edges'",
            )
            .fetch_optional(&pool)
            .await
            .unwrap();
            assert!(
                table_check.is_some(),
                "type_edges should exist after migration"
            );

            // Verify schema_version was updated to 11
            let version: (String,) =
                sqlx::query_as("SELECT value FROM metadata WHERE key = 'schema_version'")
                    .fetch_one(&pool)
                    .await
                    .unwrap();
            assert_eq!(version.0, "11");

            // Verify the indexes were created
            let idx_source: Option<(String,)> = sqlx::query_as(
                "SELECT name FROM sqlite_master WHERE type='index' AND name='idx_type_edges_source'",
            )
            .fetch_optional(&pool)
            .await
            .unwrap();
            assert!(idx_source.is_some(), "source index should exist");

            let idx_target: Option<(String,)> = sqlx::query_as(
                "SELECT name FROM sqlite_master WHERE type='index' AND name='idx_type_edges_target'",
            )
            .fetch_optional(&pool)
            .await
            .unwrap();
            assert!(idx_target.is_some(), "target index should exist");
        });
    }

    #[test]
    fn test_migrate_v12_to_v13() {
        // Full migration test: set up a v12 schema, run migration, verify enrichment_hash + hnsw_dirty
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("test.db");

        rt.block_on(async {
            let pool = SqlitePoolOptions::new()
                .max_connections(1)
                .connect_with(
                    sqlx::sqlite::SqliteConnectOptions::new()
                        .filename(&db_path)
                        .create_if_missing(true),
                )
                .await
                .unwrap();

            // Create v12 schema: chunks WITHOUT enrichment_hash, metadata WITHOUT hnsw_dirty
            sqlx::query(
                "CREATE TABLE IF NOT EXISTS metadata (
                    key TEXT PRIMARY KEY,
                    value TEXT NOT NULL
                )",
            )
            .execute(&pool)
            .await
            .unwrap();

            sqlx::query(
                "CREATE TABLE IF NOT EXISTS chunks (
                    id TEXT PRIMARY KEY,
                    origin TEXT NOT NULL,
                    source_type TEXT NOT NULL,
                    language TEXT NOT NULL,
                    chunk_type TEXT NOT NULL,
                    name TEXT NOT NULL,
                    signature TEXT NOT NULL,
                    content TEXT NOT NULL,
                    content_hash TEXT NOT NULL,
                    doc TEXT,
                    line_start INTEGER NOT NULL,
                    line_end INTEGER NOT NULL,
                    embedding BLOB NOT NULL,
                    source_mtime INTEGER,
                    created_at TEXT NOT NULL,
                    updated_at TEXT NOT NULL,
                    parent_id TEXT,
                    window_idx INTEGER,
                    parent_type_name TEXT
                )",
            )
            .execute(&pool)
            .await
            .unwrap();

            sqlx::query("INSERT INTO metadata (key, value) VALUES ('schema_version', '12')")
                .execute(&pool)
                .await
                .unwrap();

            // Run migration from v12 to v13
            let pool = migrate(pool, &db_path, 12, 13).await.unwrap();

            // Verify enrichment_hash column exists by inserting a row that uses it
            sqlx::query(
                "INSERT INTO chunks (id, origin, source_type, language, chunk_type, name, \
                 signature, content, content_hash, line_start, line_end, embedding, \
                 created_at, updated_at, enrichment_hash) \
                 VALUES ('test', 'file:test.rs', 'file', 'rust', 'function', 'test_fn', \
                 '', 'fn test() {}', 'abc123', 0, 1, X'00', '2026-01-01', '2026-01-01', 'hash123')",
            )
            .execute(&pool)
            .await
            .unwrap();

            // Verify hnsw_dirty metadata key exists with value '0'
            let dirty: (String,) =
                sqlx::query_as("SELECT value FROM metadata WHERE key = 'hnsw_dirty'")
                    .fetch_one(&pool)
                    .await
                    .unwrap();
            assert_eq!(dirty.0, "0");

            // Verify schema_version was updated to 13
            let version: (String,) =
                sqlx::query_as("SELECT value FROM metadata WHERE key = 'schema_version'")
                    .fetch_one(&pool)
                    .await
                    .unwrap();
            assert_eq!(version.0, "13");
        });
    }

    #[test]
    fn test_migrate_v13_to_v14() {
        // Full migration test: set up a v13 schema, run migration, verify llm_summaries table
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("test.db");

        rt.block_on(async {
            let pool = SqlitePoolOptions::new()
                .max_connections(1)
                .connect_with(
                    sqlx::sqlite::SqliteConnectOptions::new()
                        .filename(&db_path)
                        .create_if_missing(true),
                )
                .await
                .unwrap();

            // Create v13 schema: chunks WITH enrichment_hash, metadata WITH hnsw_dirty
            sqlx::query(
                "CREATE TABLE IF NOT EXISTS metadata (
                    key TEXT PRIMARY KEY,
                    value TEXT NOT NULL
                )",
            )
            .execute(&pool)
            .await
            .unwrap();

            sqlx::query(
                "CREATE TABLE IF NOT EXISTS chunks (
                    id TEXT PRIMARY KEY,
                    origin TEXT NOT NULL,
                    source_type TEXT NOT NULL,
                    language TEXT NOT NULL,
                    chunk_type TEXT NOT NULL,
                    name TEXT NOT NULL,
                    signature TEXT NOT NULL,
                    content TEXT NOT NULL,
                    content_hash TEXT NOT NULL,
                    doc TEXT,
                    line_start INTEGER NOT NULL,
                    line_end INTEGER NOT NULL,
                    embedding BLOB NOT NULL,
                    source_mtime INTEGER,
                    created_at TEXT NOT NULL,
                    updated_at TEXT NOT NULL,
                    parent_id TEXT,
                    window_idx INTEGER,
                    parent_type_name TEXT,
                    enrichment_hash TEXT
                )",
            )
            .execute(&pool)
            .await
            .unwrap();

            sqlx::query("INSERT INTO metadata (key, value) VALUES ('schema_version', '13')")
                .execute(&pool)
                .await
                .unwrap();
            sqlx::query("INSERT INTO metadata (key, value) VALUES ('hnsw_dirty', '0')")
                .execute(&pool)
                .await
                .unwrap();

            // Verify llm_summaries does NOT exist before migration
            let table_check: Option<(String,)> = sqlx::query_as(
                "SELECT name FROM sqlite_master WHERE type='table' AND name='llm_summaries'",
            )
            .fetch_optional(&pool)
            .await
            .unwrap();
            assert!(table_check.is_none(), "llm_summaries should not exist yet");

            // Run migration from v13 to v14
            let pool = migrate(pool, &db_path, 13, 14).await.unwrap();

            // Verify llm_summaries table exists
            let table_check: Option<(String,)> = sqlx::query_as(
                "SELECT name FROM sqlite_master WHERE type='table' AND name='llm_summaries'",
            )
            .fetch_optional(&pool)
            .await
            .unwrap();
            assert!(
                table_check.is_some(),
                "llm_summaries should exist after migration"
            );

            // Verify we can insert into llm_summaries
            sqlx::query(
                "INSERT INTO llm_summaries (content_hash, summary, model, created_at) \
                 VALUES ('abc123', 'Test summary', 'claude-4', '2026-01-01')",
            )
            .execute(&pool)
            .await
            .unwrap();

            // Verify schema_version was updated to 14
            let version: (String,) =
                sqlx::query_as("SELECT value FROM metadata WHERE key = 'schema_version'")
                    .fetch_one(&pool)
                    .await
                    .unwrap();
            assert_eq!(version.0, "14");
        });
    }

    #[test]
    fn test_migrate_v14_to_v15() {
        // Full migration test: set up a v14 schema, run migration, verify dimensions + hnsw_dirty
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("test.db");

        rt.block_on(async {
            let pool = SqlitePoolOptions::new()
                .max_connections(1)
                .connect_with(
                    sqlx::sqlite::SqliteConnectOptions::new()
                        .filename(&db_path)
                        .create_if_missing(true),
                )
                .await
                .unwrap();

            // Create v14 schema: chunks WITH enrichment_hash, llm_summaries table
            sqlx::query(
                "CREATE TABLE IF NOT EXISTS metadata (
                    key TEXT PRIMARY KEY,
                    value TEXT NOT NULL
                )",
            )
            .execute(&pool)
            .await
            .unwrap();

            sqlx::query(
                "CREATE TABLE IF NOT EXISTS chunks (
                    id TEXT PRIMARY KEY,
                    origin TEXT NOT NULL,
                    source_type TEXT NOT NULL,
                    language TEXT NOT NULL,
                    chunk_type TEXT NOT NULL,
                    name TEXT NOT NULL,
                    signature TEXT NOT NULL,
                    content TEXT NOT NULL,
                    content_hash TEXT NOT NULL,
                    doc TEXT,
                    line_start INTEGER NOT NULL,
                    line_end INTEGER NOT NULL,
                    embedding BLOB NOT NULL,
                    source_mtime INTEGER,
                    created_at TEXT NOT NULL,
                    updated_at TEXT NOT NULL,
                    parent_id TEXT,
                    window_idx INTEGER,
                    parent_type_name TEXT,
                    enrichment_hash TEXT
                )",
            )
            .execute(&pool)
            .await
            .unwrap();

            sqlx::query(
                "CREATE TABLE IF NOT EXISTS llm_summaries (
                    content_hash TEXT PRIMARY KEY,
                    summary TEXT NOT NULL,
                    model TEXT NOT NULL,
                    created_at TEXT NOT NULL
                )",
            )
            .execute(&pool)
            .await
            .unwrap();

            sqlx::query("INSERT INTO metadata (key, value) VALUES ('schema_version', '14')")
                .execute(&pool)
                .await
                .unwrap();
            sqlx::query("INSERT INTO metadata (key, value) VALUES ('dimensions', '769')")
                .execute(&pool)
                .await
                .unwrap();
            sqlx::query("INSERT INTO metadata (key, value) VALUES ('hnsw_dirty', '0')")
                .execute(&pool)
                .await
                .unwrap();

            // Run migration from v14 to v15
            let pool = migrate(pool, &db_path, 14, 15).await.unwrap();

            // Verify dimensions updated to 768
            let dims: (String,) =
                sqlx::query_as("SELECT value FROM metadata WHERE key = 'dimensions'")
                    .fetch_one(&pool)
                    .await
                    .unwrap();
            assert_eq!(dims.0, "768", "dimensions should be updated to 768");

            // Verify hnsw_dirty set to 1 (triggers rebuild)
            let dirty: (String,) =
                sqlx::query_as("SELECT value FROM metadata WHERE key = 'hnsw_dirty'")
                    .fetch_one(&pool)
                    .await
                    .unwrap();
            assert_eq!(dirty.0, "1", "hnsw_dirty should be set to 1");

            // Verify schema_version was updated to 15
            let version: (String,) =
                sqlx::query_as("SELECT value FROM metadata WHERE key = 'schema_version'")
                    .fetch_one(&pool)
                    .await
                    .unwrap();
            assert_eq!(version.0, "15");
        });
    }

    #[test]
    fn test_migrate_v15_to_v16() {
        // Full migration test: set up a v15 schema, run migration, verify composite PK
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("test.db");

        rt.block_on(async {
            let pool = SqlitePoolOptions::new()
                .max_connections(1)
                .connect_with(
                    sqlx::sqlite::SqliteConnectOptions::new()
                        .filename(&db_path)
                        .create_if_missing(true),
                )
                .await
                .unwrap();

            // Create v15 schema with llm_summaries (single PK on content_hash)
            sqlx::query(
                "CREATE TABLE IF NOT EXISTS metadata (
                    key TEXT PRIMARY KEY,
                    value TEXT NOT NULL
                )",
            )
            .execute(&pool)
            .await
            .unwrap();

            sqlx::query(
                "CREATE TABLE IF NOT EXISTS llm_summaries (
                    content_hash TEXT PRIMARY KEY,
                    summary TEXT NOT NULL,
                    model TEXT NOT NULL,
                    created_at TEXT NOT NULL
                )",
            )
            .execute(&pool)
            .await
            .unwrap();

            sqlx::query("INSERT INTO metadata (key, value) VALUES ('schema_version', '15')")
                .execute(&pool)
                .await
                .unwrap();

            // Insert two test summaries
            sqlx::query(
                "INSERT INTO llm_summaries (content_hash, summary, model, created_at) \
                 VALUES ('hash_a', 'Summary A', 'claude-4', '2026-01-01')",
            )
            .execute(&pool)
            .await
            .unwrap();
            sqlx::query(
                "INSERT INTO llm_summaries (content_hash, summary, model, created_at) \
                 VALUES ('hash_b', 'Summary B', 'claude-4', '2026-01-02')",
            )
            .execute(&pool)
            .await
            .unwrap();

            // Run migration from v15 to v16
            let pool = migrate(pool, &db_path, 15, 16).await.unwrap();

            // Verify existing rows have purpose='summary'
            let count: (i64,) =
                sqlx::query_as("SELECT COUNT(*) FROM llm_summaries WHERE purpose = 'summary'")
                    .fetch_one(&pool)
                    .await
                    .unwrap();
            assert_eq!(
                count.0, 2,
                "both existing rows should have purpose='summary'"
            );

            // Verify composite PK: same content_hash with different purpose should succeed
            sqlx::query(
                "INSERT INTO llm_summaries (content_hash, purpose, summary, model, created_at) \
                 VALUES ('hash_a', 'doc-comment', 'Doc comment A', 'claude-4', '2026-01-03')",
            )
            .execute(&pool)
            .await
            .expect("inserting same content_hash with different purpose should succeed");

            // Verify we now have 3 rows total
            let count: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM llm_summaries")
                .fetch_one(&pool)
                .await
                .unwrap();
            assert_eq!(count.0, 3, "should have 3 rows after inserting doc-comment");

            // Verify schema_version was updated to 16
            let version: (String,) =
                sqlx::query_as("SELECT value FROM metadata WHERE key = 'schema_version'")
                    .fetch_one(&pool)
                    .await
                    .unwrap();
            assert_eq!(version.0, "16");
        });
    }

    #[test]
    fn test_migrate_v12_to_v14_full_chain() {
        // Full chain migration: v12 → v13 → v14 in one call
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("test.db");

        rt.block_on(async {
            let pool = SqlitePoolOptions::new()
                .max_connections(1)
                .connect_with(
                    sqlx::sqlite::SqliteConnectOptions::new()
                        .filename(&db_path)
                        .create_if_missing(true),
                )
                .await
                .unwrap();

            // Create v12 schema: chunks WITHOUT enrichment_hash, no hnsw_dirty
            sqlx::query(
                "CREATE TABLE IF NOT EXISTS metadata (
                    key TEXT PRIMARY KEY,
                    value TEXT NOT NULL
                )",
            )
            .execute(&pool)
            .await
            .unwrap();

            sqlx::query(
                "CREATE TABLE IF NOT EXISTS chunks (
                    id TEXT PRIMARY KEY,
                    origin TEXT NOT NULL,
                    source_type TEXT NOT NULL,
                    language TEXT NOT NULL,
                    chunk_type TEXT NOT NULL,
                    name TEXT NOT NULL,
                    signature TEXT NOT NULL,
                    content TEXT NOT NULL,
                    content_hash TEXT NOT NULL,
                    doc TEXT,
                    line_start INTEGER NOT NULL,
                    line_end INTEGER NOT NULL,
                    embedding BLOB NOT NULL,
                    source_mtime INTEGER,
                    created_at TEXT NOT NULL,
                    updated_at TEXT NOT NULL,
                    parent_id TEXT,
                    window_idx INTEGER,
                    parent_type_name TEXT
                )",
            )
            .execute(&pool)
            .await
            .unwrap();

            sqlx::query("INSERT INTO metadata (key, value) VALUES ('schema_version', '12')")
                .execute(&pool)
                .await
                .unwrap();

            // Run full chain migration from v12 to v14
            let pool = migrate(pool, &db_path, 12, 14).await.unwrap();

            // Verify enrichment_hash column exists (from v12→v13)
            sqlx::query(
                "INSERT INTO chunks (id, origin, source_type, language, chunk_type, name, \
                 signature, content, content_hash, line_start, line_end, embedding, \
                 created_at, updated_at, enrichment_hash) \
                 VALUES ('test', 'file:test.rs', 'file', 'rust', 'function', 'test_fn', \
                 '', 'fn test() {}', 'abc123', 0, 1, X'00', '2026-01-01', '2026-01-01', 'hash123')",
            )
            .execute(&pool)
            .await
            .unwrap();

            // Verify llm_summaries table exists (from v13→v14)
            let table_check: Option<(String,)> = sqlx::query_as(
                "SELECT name FROM sqlite_master WHERE type='table' AND name='llm_summaries'",
            )
            .fetch_optional(&pool)
            .await
            .unwrap();
            assert!(
                table_check.is_some(),
                "llm_summaries should exist after full chain migration"
            );

            // Verify schema_version was updated to 14
            let version: (String,) =
                sqlx::query_as("SELECT value FROM metadata WHERE key = 'schema_version'")
                    .fetch_one(&pool)
                    .await
                    .unwrap();
            assert_eq!(version.0, "14");
        });
    }

    #[test]
    fn test_migrate_unsupported_version_range() {
        // Migration from an unsupported range should fail with MigrationNotSupported
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("test.db");

        rt.block_on(async {
            let pool = SqlitePoolOptions::new()
                .max_connections(1)
                .connect_with(
                    sqlx::sqlite::SqliteConnectOptions::new()
                        .filename(&db_path)
                        .create_if_missing(true),
                )
                .await
                .unwrap();

            // Create metadata table so the SQL doesn't fail on table-not-found
            sqlx::query(
                "CREATE TABLE IF NOT EXISTS metadata (
                    key TEXT PRIMARY KEY,
                    value TEXT NOT NULL
                )",
            )
            .execute(&pool)
            .await
            .unwrap();

            sqlx::query("INSERT INTO metadata (key, value) VALUES ('schema_version', '8')")
                .execute(&pool)
                .await
                .unwrap();

            let result = migrate(pool, &db_path, 8, 11).await;
            assert!(result.is_err(), "unsupported range should fail");
            match result.unwrap_err() {
                StoreError::MigrationNotSupported { from, to } => {
                    assert_eq!(from, 8);
                    assert_eq!(to, 9);
                }
                other => panic!("Expected MigrationNotSupported, got: {:?}", other),
            }
        });
    }

    /// v17→v18 adds embedding_base column without touching existing rows, and
    /// the migration is idempotent-ish in the sense that a follow-up attempt
    /// errors on the duplicate ALTER (caller must not re-run).
    #[test]
    fn test_migrate_v17_to_v18_adds_embedding_base_column() {
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("test.db");

        rt.block_on(async {
            let pool = SqlitePoolOptions::new()
                .max_connections(1)
                .connect_with(
                    sqlx::sqlite::SqliteConnectOptions::new()
                        .filename(&db_path)
                        .create_if_missing(true),
                )
                .await
                .unwrap();

            // Minimal v17 schema (chunks + metadata); only the columns
            // touched by the v17→v18 migration matter here.
            sqlx::query("CREATE TABLE metadata (key TEXT PRIMARY KEY, value TEXT NOT NULL)")
                .execute(&pool)
                .await
                .unwrap();
            sqlx::query(
                "CREATE TABLE chunks (
                    id TEXT PRIMARY KEY,
                    origin TEXT NOT NULL,
                    source_type TEXT NOT NULL,
                    language TEXT NOT NULL,
                    chunk_type TEXT NOT NULL,
                    name TEXT NOT NULL,
                    signature TEXT NOT NULL,
                    content TEXT NOT NULL,
                    content_hash TEXT NOT NULL,
                    line_start INTEGER NOT NULL,
                    line_end INTEGER NOT NULL,
                    embedding BLOB NOT NULL,
                    created_at TEXT NOT NULL,
                    updated_at TEXT NOT NULL,
                    enrichment_version INTEGER NOT NULL DEFAULT 0
                )",
            )
            .execute(&pool)
            .await
            .unwrap();

            sqlx::query("INSERT INTO metadata (key, value) VALUES ('schema_version', '17')")
                .execute(&pool)
                .await
                .unwrap();

            // Insert a row with a non-trivial embedding so we can verify it
            // survives the migration untouched.
            sqlx::query(
                "INSERT INTO chunks (id, origin, source_type, language, chunk_type, name, \
                 signature, content, content_hash, line_start, line_end, embedding, \
                 created_at, updated_at) \
                 VALUES ('chunk-1', 'file:lib.rs', 'file', 'rust', 'function', 'foo', \
                 'fn foo()', 'fn foo() {}', 'hash1', 10, 20, X'deadbeef', \
                 '2026-04-10', '2026-04-10')",
            )
            .execute(&pool)
            .await
            .unwrap();

            let pool = migrate(pool, &db_path, 17, 18).await.unwrap();

            // Column exists and defaults to NULL for pre-existing rows.
            let (embedding_existing, embedding_base): (Vec<u8>, Option<Vec<u8>>) =
                sqlx::query_as("SELECT embedding, embedding_base FROM chunks WHERE id = 'chunk-1'")
                    .fetch_one(&pool)
                    .await
                    .unwrap();
            assert_eq!(
                embedding_existing,
                vec![0xde, 0xad, 0xbe, 0xef],
                "existing embedding must survive migration untouched"
            );
            assert!(
                embedding_base.is_none(),
                "embedding_base must be NULL for pre-existing rows (base pass hasn't run yet)"
            );

            // Schema version bumped.
            let version: (String,) =
                sqlx::query_as("SELECT value FROM metadata WHERE key = 'schema_version'")
                    .fetch_one(&pool)
                    .await
                    .unwrap();
            assert_eq!(version.0, "18");

            // NULL is a writeable state — caller can populate it later.
            sqlx::query("UPDATE chunks SET embedding_base = X'cafef00d' WHERE id = 'chunk-1'")
                .execute(&pool)
                .await
                .unwrap();
            let (base_after,): (Option<Vec<u8>>,) =
                sqlx::query_as("SELECT embedding_base FROM chunks WHERE id = 'chunk-1'")
                    .fetch_one(&pool)
                    .await
                    .unwrap();
            assert_eq!(base_after, Some(vec![0xca, 0xfe, 0xf0, 0x0d]));
        });
    }

    /// Full migrate() chain is idempotent at the dispatcher level.
    /// Calling `migrate(pool, 18, 18)` after a successful upgrade must be a
    /// no-op — the schema_version metadata gates re-execution of the ALTER.
    /// (The raw migration function itself is NOT idempotent — `ALTER TABLE
    /// ADD COLUMN` errors on duplicate column. This test exercises the
    /// dispatcher contract that protects users from the underlying limitation.)
    #[test]
    fn test_migrate_v17_to_v18_dispatcher_is_idempotent() {
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("test.db");

        rt.block_on(async {
            let pool = SqlitePoolOptions::new()
                .max_connections(1)
                .connect_with(
                    sqlx::sqlite::SqliteConnectOptions::new()
                        .filename(&db_path)
                        .create_if_missing(true),
                )
                .await
                .unwrap();

            sqlx::query("CREATE TABLE metadata (key TEXT PRIMARY KEY, value TEXT NOT NULL)")
                .execute(&pool)
                .await
                .unwrap();
            sqlx::query(
                "CREATE TABLE chunks (
                    id TEXT PRIMARY KEY,
                    origin TEXT NOT NULL,
                    source_type TEXT NOT NULL,
                    language TEXT NOT NULL,
                    chunk_type TEXT NOT NULL,
                    name TEXT NOT NULL,
                    signature TEXT NOT NULL,
                    content TEXT NOT NULL,
                    content_hash TEXT NOT NULL,
                    line_start INTEGER NOT NULL,
                    line_end INTEGER NOT NULL,
                    embedding BLOB NOT NULL,
                    created_at TEXT NOT NULL,
                    updated_at TEXT NOT NULL,
                    enrichment_version INTEGER NOT NULL DEFAULT 0
                )",
            )
            .execute(&pool)
            .await
            .unwrap();
            sqlx::query("INSERT INTO metadata (key, value) VALUES ('schema_version', '17')")
                .execute(&pool)
                .await
                .unwrap();

            // First call: 17→18 succeeds.
            let pool = migrate(pool, &db_path, 17, 18).await.unwrap();

            // Second call at the same target version: should be a no-op.
            // This is the property users actually depend on — re-running
            // `cqs index` should not fail just because the schema is current.
            let _pool = migrate(pool, &db_path, 18, 18).await.unwrap();
        });
    }

    /// v18→v19 migration adds FK(chunk_id) ON DELETE CASCADE on
    /// sparse_vectors. Test covers:
    /// - Orphan sparse rows (chunks no longer exist) are dropped during
    ///   the rebuild
    /// - Non-orphan sparse rows survive the rebuild
    /// - The idx_sparse_token index is recreated
    /// - splade_generation is bumped so any pre-v19 persisted index file
    ///   is invalidated
    /// - FK CASCADE works after the migration: deleting a chunk auto-deletes
    ///   its sparse rows
    #[test]
    fn test_migrate_v18_to_v19_adds_fk_cascade_and_purges_orphans() {
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("test.db");

        rt.block_on(async {
            let pool = SqlitePoolOptions::new()
                .max_connections(1)
                .connect_with(
                    sqlx::sqlite::SqliteConnectOptions::new()
                        .filename(&db_path)
                        .create_if_missing(true)
                        // PRAGMA foreign_keys must be ON for ON DELETE CASCADE to fire.
                        .foreign_keys(true),
                )
                .await
                .unwrap();

            // Minimal v18 schema — chunks with embedding_base, sparse_vectors
            // WITHOUT the FK constraint (as shipped in v17).
            sqlx::query("CREATE TABLE metadata (key TEXT PRIMARY KEY, value TEXT NOT NULL)")
                .execute(&pool)
                .await
                .unwrap();
            sqlx::query(
                "CREATE TABLE chunks (
                    id TEXT PRIMARY KEY,
                    origin TEXT NOT NULL,
                    source_type TEXT NOT NULL,
                    language TEXT NOT NULL,
                    chunk_type TEXT NOT NULL,
                    name TEXT NOT NULL,
                    signature TEXT NOT NULL,
                    content TEXT NOT NULL,
                    content_hash TEXT NOT NULL,
                    line_start INTEGER NOT NULL,
                    line_end INTEGER NOT NULL,
                    embedding BLOB NOT NULL,
                    embedding_base BLOB,
                    created_at TEXT NOT NULL,
                    updated_at TEXT NOT NULL,
                    enrichment_version INTEGER NOT NULL DEFAULT 0
                )",
            )
            .execute(&pool)
            .await
            .unwrap();
            sqlx::query(
                "CREATE TABLE sparse_vectors (
                    chunk_id TEXT NOT NULL,
                    token_id INTEGER NOT NULL,
                    weight REAL NOT NULL,
                    PRIMARY KEY (chunk_id, token_id)
                )",
            )
            .execute(&pool)
            .await
            .unwrap();
            sqlx::query("CREATE INDEX idx_sparse_token ON sparse_vectors(token_id)")
                .execute(&pool)
                .await
                .unwrap();

            sqlx::query(
                "INSERT INTO metadata (key, value) VALUES ('schema_version', '18'),
                                                          ('splade_generation', '5')",
            )
            .execute(&pool)
            .await
            .unwrap();

            // Insert two chunks and sparse rows for both of them.
            for id in ["chunk-live", "chunk-also-live"] {
                sqlx::query(
                    "INSERT INTO chunks (id, origin, source_type, language, chunk_type, name, \
                     signature, content, content_hash, line_start, line_end, embedding, \
                     created_at, updated_at) \
                     VALUES (?1, 'file:lib.rs', 'file', 'rust', 'function', ?1, \
                     '', '', 'hash', 1, 10, X'00', '2026-04-11', '2026-04-11')",
                )
                .bind(id)
                .execute(&pool)
                .await
                .unwrap();
            }
            sqlx::query(
                "INSERT INTO sparse_vectors (chunk_id, token_id, weight) VALUES
                    ('chunk-live', 1, 0.5),
                    ('chunk-live', 2, 0.3),
                    ('chunk-also-live', 3, 0.8),
                    ('chunk-orphan', 4, 0.9),
                    ('chunk-orphan', 5, 0.1)",
            )
            .execute(&pool)
            .await
            .unwrap();

            // Run the migration.
            let pool = migrate(pool, &db_path, 18, 19).await.unwrap();

            // Orphan rows dropped, live rows survive.
            let (count,): (i64,) = sqlx::query_as(
                "SELECT COUNT(*) FROM sparse_vectors WHERE chunk_id = 'chunk-orphan'",
            )
            .fetch_one(&pool)
            .await
            .unwrap();
            assert_eq!(count, 0, "orphan rows should have been dropped");
            let (count_live,): (i64,) =
                sqlx::query_as("SELECT COUNT(*) FROM sparse_vectors WHERE chunk_id = 'chunk-live'")
                    .fetch_one(&pool)
                    .await
                    .unwrap();
            assert_eq!(count_live, 2, "chunk-live rows should survive");

            // idx_sparse_token exists after the rebuild.
            let idx: Option<(String,)> = sqlx::query_as(
                "SELECT name FROM sqlite_master WHERE type='index' AND name='idx_sparse_token'",
            )
            .fetch_optional(&pool)
            .await
            .unwrap();
            assert!(idx.is_some(), "idx_sparse_token must be recreated");

            // splade_generation bumped.
            let (gen_val,): (String,) =
                sqlx::query_as("SELECT value FROM metadata WHERE key = 'splade_generation'")
                    .fetch_one(&pool)
                    .await
                    .unwrap();
            let gen_parsed: u64 = gen_val.parse().unwrap();
            assert!(gen_parsed > 5, "splade_generation must be bumped past 5");

            // schema_version updated.
            let (v,): (String,) =
                sqlx::query_as("SELECT value FROM metadata WHERE key = 'schema_version'")
                    .fetch_one(&pool)
                    .await
                    .unwrap();
            assert_eq!(v, "19");

            // FK CASCADE contract: deleting a chunk removes its sparse rows.
            sqlx::query("DELETE FROM chunks WHERE id = 'chunk-also-live'")
                .execute(&pool)
                .await
                .unwrap();
            let (remaining,): (i64,) = sqlx::query_as(
                "SELECT COUNT(*) FROM sparse_vectors WHERE chunk_id = 'chunk-also-live'",
            )
            .fetch_one(&pool)
            .await
            .unwrap();
            assert_eq!(
                remaining, 0,
                "CASCADE should have removed chunk-also-live's sparse rows"
            );
        });
    }

    /// v19→v20 adds a trigger on chunks that bumps splade_generation on
    /// every DELETE. Test covers:
    /// - Migration bumps the generation immediately
    /// - The trigger fires on subsequent DELETE FROM chunks
    /// - Multiple deletes produce multiple bumps (cardinality check)
    /// - A no-op reindex (all INSERT no DELETE) does NOT bump via the trigger
    #[test]
    fn test_migrate_v19_to_v20_adds_trigger_that_bumps_on_chunks_delete() {
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("test.db");

        rt.block_on(async {
            let pool = SqlitePoolOptions::new()
                .max_connections(1)
                .connect_with(
                    sqlx::sqlite::SqliteConnectOptions::new()
                        .filename(&db_path)
                        .create_if_missing(true)
                        .foreign_keys(true),
                )
                .await
                .unwrap();

            // Minimal v19 schema: chunks, sparse_vectors with v19 FK, metadata.
            sqlx::query("CREATE TABLE metadata (key TEXT PRIMARY KEY, value TEXT NOT NULL)")
                .execute(&pool)
                .await
                .unwrap();
            sqlx::query(
                "CREATE TABLE chunks (
                    id TEXT PRIMARY KEY,
                    origin TEXT NOT NULL,
                    source_type TEXT NOT NULL,
                    language TEXT NOT NULL,
                    chunk_type TEXT NOT NULL,
                    name TEXT NOT NULL,
                    signature TEXT NOT NULL,
                    content TEXT NOT NULL,
                    content_hash TEXT NOT NULL,
                    line_start INTEGER NOT NULL,
                    line_end INTEGER NOT NULL,
                    embedding BLOB NOT NULL,
                    embedding_base BLOB,
                    created_at TEXT NOT NULL,
                    updated_at TEXT NOT NULL,
                    enrichment_version INTEGER NOT NULL DEFAULT 0
                )",
            )
            .execute(&pool)
            .await
            .unwrap();
            sqlx::query(
                "CREATE TABLE sparse_vectors (
                    chunk_id TEXT NOT NULL,
                    token_id INTEGER NOT NULL,
                    weight REAL NOT NULL,
                    PRIMARY KEY (chunk_id, token_id),
                    FOREIGN KEY (chunk_id) REFERENCES chunks(id) ON DELETE CASCADE
                )",
            )
            .execute(&pool)
            .await
            .unwrap();
            sqlx::query("CREATE INDEX idx_sparse_token ON sparse_vectors(token_id)")
                .execute(&pool)
                .await
                .unwrap();
            sqlx::query(
                "INSERT INTO metadata (key, value) VALUES ('schema_version', '19'),
                                                          ('splade_generation', '10')",
            )
            .execute(&pool)
            .await
            .unwrap();

            // Seed a chunk so we have something to delete.
            for id in ["c1", "c2", "c3"] {
                sqlx::query(
                    "INSERT INTO chunks (id, origin, source_type, language, chunk_type, name, \
                     signature, content, content_hash, line_start, line_end, embedding, \
                     created_at, updated_at) \
                     VALUES (?1, 'file:lib.rs', 'file', 'rust', 'function', ?1, \
                     '', '', 'h', 1, 10, X'00', '2026-04-12', '2026-04-12')",
                )
                .bind(id)
                .execute(&pool)
                .await
                .unwrap();
            }

            // Run v19 → v20 migration.
            let pool = migrate(pool, &db_path, 19, 20).await.unwrap();

            // Migration itself bumps the generation once.
            let (gen_after_migration,): (String,) =
                sqlx::query_as("SELECT value FROM metadata WHERE key = 'splade_generation'")
                    .fetch_one(&pool)
                    .await
                    .unwrap();
            let gen_after_migration: u64 = gen_after_migration.parse().unwrap();
            assert!(
                gen_after_migration > 10,
                "migration should bump generation past starting value 10"
            );

            // Trigger exists.
            let trigger: Option<(String,)> = sqlx::query_as(
                "SELECT name FROM sqlite_master WHERE type='trigger' \
                 AND name='bump_splade_on_chunks_delete'",
            )
            .fetch_optional(&pool)
            .await
            .unwrap();
            assert!(trigger.is_some(), "v20 trigger must exist after migration");

            // Schema version.
            let (v,): (String,) =
                sqlx::query_as("SELECT value FROM metadata WHERE key = 'schema_version'")
                    .fetch_one(&pool)
                    .await
                    .unwrap();
            assert_eq!(v, "20");

            // Deleting a chunk fires the trigger once and bumps the
            // generation.
            let before_one_delete = gen_after_migration;
            sqlx::query("DELETE FROM chunks WHERE id = 'c1'")
                .execute(&pool)
                .await
                .unwrap();
            let (gen_after_one_delete,): (String,) =
                sqlx::query_as("SELECT value FROM metadata WHERE key = 'splade_generation'")
                    .fetch_one(&pool)
                    .await
                    .unwrap();
            let gen_after_one_delete: u64 = gen_after_one_delete.parse().unwrap();
            assert_eq!(
                gen_after_one_delete,
                before_one_delete + 1,
                "one chunk delete should bump generation by exactly one"
            );

            // Deleting two chunks bumps by two (trigger is row-level).
            sqlx::query("DELETE FROM chunks WHERE id IN ('c2', 'c3')")
                .execute(&pool)
                .await
                .unwrap();
            let (gen_after_two_delete,): (String,) =
                sqlx::query_as("SELECT value FROM metadata WHERE key = 'splade_generation'")
                    .fetch_one(&pool)
                    .await
                    .unwrap();
            let gen_after_two_delete: u64 = gen_after_two_delete.parse().unwrap();
            assert_eq!(
                gen_after_two_delete,
                gen_after_one_delete + 2,
                "two chunk deletes should bump generation by exactly two"
            );

            // INSERTs do NOT bump via this trigger (new chunks don't
            // invalidate existing sparse data).
            sqlx::query(
                "INSERT INTO chunks (id, origin, source_type, language, chunk_type, name, \
                 signature, content, content_hash, line_start, line_end, embedding, \
                 created_at, updated_at) \
                 VALUES ('c4', 'file:lib.rs', 'file', 'rust', 'function', 'c4', \
                 '', '', 'h', 1, 10, X'00', '2026-04-12', '2026-04-12')",
            )
            .execute(&pool)
            .await
            .unwrap();
            let (gen_after_insert,): (String,) =
                sqlx::query_as("SELECT value FROM metadata WHERE key = 'splade_generation'")
                    .fetch_one(&pool)
                    .await
                    .unwrap();
            let gen_after_insert: u64 = gen_after_insert.parse().unwrap();
            assert_eq!(
                gen_after_insert, gen_after_two_delete,
                "INSERT should NOT bump the generation (no DELETE happened)"
            );
        });
    }

    // ========================================================================
    // Filesystem backup before migrate() runs DDL.
    //
    // The four tests below cover the happy-path backup creation, the
    // failure-path restore, the pruning policy, and the absent-WAL edge case.
    // ========================================================================

    /// Helper: build a minimal v19 schema on `pool` so migrate(19 -> 20) can
    /// run against it. Each backup test uses this so the test body focuses on
    /// backup/restore behaviour rather than schema setup.
    async fn seed_v19_schema(pool: &sqlx::SqlitePool) {
        sqlx::query("CREATE TABLE metadata (key TEXT PRIMARY KEY, value TEXT NOT NULL)")
            .execute(pool)
            .await
            .unwrap();
        sqlx::query(
            "CREATE TABLE chunks (
                id TEXT PRIMARY KEY,
                origin TEXT NOT NULL,
                source_type TEXT NOT NULL,
                language TEXT NOT NULL,
                chunk_type TEXT NOT NULL,
                name TEXT NOT NULL,
                signature TEXT NOT NULL,
                content TEXT NOT NULL,
                content_hash TEXT NOT NULL,
                line_start INTEGER NOT NULL,
                line_end INTEGER NOT NULL,
                embedding BLOB NOT NULL,
                embedding_base BLOB,
                created_at TEXT NOT NULL,
                updated_at TEXT NOT NULL,
                enrichment_version INTEGER NOT NULL DEFAULT 0
            )",
        )
        .execute(pool)
        .await
        .unwrap();
        sqlx::query(
            "CREATE TABLE sparse_vectors (
                chunk_id TEXT NOT NULL,
                token_id INTEGER NOT NULL,
                weight REAL NOT NULL,
                PRIMARY KEY (chunk_id, token_id),
                FOREIGN KEY (chunk_id) REFERENCES chunks(id) ON DELETE CASCADE
            )",
        )
        .execute(pool)
        .await
        .unwrap();
        sqlx::query("CREATE INDEX idx_sparse_token ON sparse_vectors(token_id)")
            .execute(pool)
            .await
            .unwrap();
        sqlx::query(
            "INSERT INTO metadata (key, value) VALUES ('schema_version', '19'),
                                                      ('splade_generation', '7')",
        )
        .execute(pool)
        .await
        .unwrap();
    }

    /// Happy path: a successful migrate() leaves a
    /// `.bak-v{from}-v{to}-{ts}.db` file in the DB's parent directory.
    ///
    /// This is what gives users a recovery path when a future migration
    /// fails on real data — they can `mv` the backup back into place
    /// instead of re-indexing their corpus from source.
    #[test]
    fn test_migrate_writes_backup_file_on_success() {
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("test.db");

        rt.block_on(async {
            let pool = SqlitePoolOptions::new()
                .max_connections(1)
                .connect_with(
                    sqlx::sqlite::SqliteConnectOptions::new()
                        .filename(&db_path)
                        .create_if_missing(true)
                        .foreign_keys(true),
                )
                .await
                .unwrap();
            seed_v19_schema(&pool).await;

            // Seed one chunk so the DB is not empty — lets us catch a bug
            // where the backup captures a zero-length file instead of the
            // live one.
            sqlx::query(
                "INSERT INTO chunks (id, origin, source_type, language, chunk_type, name, \
                 signature, content, content_hash, line_start, line_end, embedding, \
                 created_at, updated_at) \
                 VALUES ('c1', 'file:lib.rs', 'file', 'rust', 'function', 'c1', \
                 '', '', 'h', 1, 10, X'00', '2026-04-15', '2026-04-15')",
            )
            .execute(&pool)
            .await
            .unwrap();

            let _pool = migrate(pool, &db_path, 19, 20).await.unwrap();

            // Enumerate backups — exactly one should exist after this run.
            let backups: Vec<_> = std::fs::read_dir(dir.path())
                .unwrap()
                .flatten()
                .filter_map(|e| e.file_name().to_str().map(|s| s.to_string()))
                .filter(|n| n.starts_with("test.bak-v19-v20-") && n.ends_with(".db"))
                .collect();
            assert_eq!(
                backups.len(),
                1,
                "expected exactly one .bak-v19-v20-*.db file, got: {:?}",
                backups
            );

            // Backup should be non-empty (not a placeholder for a failed copy).
            let bak_path = dir.path().join(&backups[0]);
            let bak_bytes = std::fs::read(&bak_path).unwrap();
            assert!(
                !bak_bytes.is_empty(),
                "backup DB file should not be zero-length"
            );
            // SQLite files start with the 16-byte header "SQLite format 3\0".
            assert!(
                bak_bytes.starts_with(b"SQLite format 3\0"),
                "backup should be a valid SQLite database file"
            );
        });
    }

    /// Failure path: when a migration step returns `Err` mid-way
    /// (simulating either a bug inside a migration function or a
    /// commit-time I/O failure), the restore-from-backup path runs and the
    /// DB file is byte-identical to its pre-migrate state.
    ///
    /// Uses the thread-local `TEST_FAIL_AFTER_VERSION` hook (gated on
    /// `cfg(test)` inside `run_migration_tx`) to trigger the failure
    /// deterministically. Thread-local so parallel v17→v18 tests on other
    /// threads don't see the injected failure.
    #[test]
    fn test_migrate_restores_db_on_failure() {
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("test.db");

        // Phase 1: set up a v17 DB, close the pool so SQLite checkpoints
        // and removes the WAL, then snapshot the main DB file bytes. These
        // bytes are what restore must reproduce.
        rt.block_on(async {
            let pool = SqlitePoolOptions::new()
                .max_connections(1)
                .connect_with(
                    sqlx::sqlite::SqliteConnectOptions::new()
                        .filename(&db_path)
                        .create_if_missing(true)
                        .journal_mode(sqlx::sqlite::SqliteJournalMode::Wal)
                        .synchronous(sqlx::sqlite::SqliteSynchronous::Full),
                )
                .await
                .unwrap();
            sqlx::query("CREATE TABLE metadata (key TEXT PRIMARY KEY, value TEXT NOT NULL)")
                .execute(&pool)
                .await
                .unwrap();
            sqlx::query(
                "CREATE TABLE chunks (
                    id TEXT PRIMARY KEY,
                    origin TEXT NOT NULL,
                    source_type TEXT NOT NULL,
                    language TEXT NOT NULL,
                    chunk_type TEXT NOT NULL,
                    name TEXT NOT NULL,
                    signature TEXT NOT NULL,
                    content TEXT NOT NULL,
                    content_hash TEXT NOT NULL,
                    line_start INTEGER NOT NULL,
                    line_end INTEGER NOT NULL,
                    embedding BLOB NOT NULL,
                    created_at TEXT NOT NULL,
                    updated_at TEXT NOT NULL,
                    enrichment_version INTEGER NOT NULL DEFAULT 0
                )",
            )
            .execute(&pool)
            .await
            .unwrap();
            sqlx::query("INSERT INTO metadata (key, value) VALUES ('schema_version', '17')")
                .execute(&pool)
                .await
                .unwrap();
            // Checkpoint and close so the WAL is drained into the main DB.
            // After this, the on-disk bytes capture the full pre-migrate state.
            sqlx::query("PRAGMA wal_checkpoint(TRUNCATE)")
                .execute(&pool)
                .await
                .unwrap();
            pool.close().await;
        });

        // Phase 2: reopen, set the failure hook, run migrate(17 -> 19).
        // The hook makes step v18 succeed then fail the overall migration.
        // On error, restore should put the DB file back to pre_migrate_bytes.
        TEST_FAIL_AFTER_VERSION.with(|c| c.set(18));

        rt.block_on(async {
            let pool = SqlitePoolOptions::new()
                .max_connections(1)
                .connect_with(
                    sqlx::sqlite::SqliteConnectOptions::new()
                        .filename(&db_path)
                        .create_if_missing(false)
                        .journal_mode(sqlx::sqlite::SqliteJournalMode::Wal)
                        .synchronous(sqlx::sqlite::SqliteSynchronous::Full),
                )
                .await
                .unwrap();

            // migrate consumes the pool by value. On the failure path it
            // closes the pool internally before restore runs, so we don't
            // need a separate close here.
            let err = migrate(pool, &db_path, 17, 19).await.unwrap_err();
            match err {
                StoreError::Runtime(msg) => assert!(
                    msg.contains("injected failure"),
                    "expected injected-failure error, got: {}",
                    msg
                ),
                other => panic!("expected Runtime(injected failure), got: {:?}", other),
            }
        });

        TEST_FAIL_AFTER_VERSION.with(|c| c.set(0));

        // The DB must be back at its pre-migrate LOGICAL state. The backup is
        // now a `VACUUM INTO` snapshot — logically identical to the source but
        // not byte-identical (VACUUM defragments page layout), so we assert on
        // content rather than raw bytes: schema_version is 17 and the column
        // that migrate_v17_to_v18 would have added (`embedding_base`) is
        // absent. If restore didn't fire, the ALTER would have added the
        // column and bumped the version.
        rt.block_on(async {
            let pool = SqlitePoolOptions::new()
                .max_connections(1)
                .connect_with(
                    sqlx::sqlite::SqliteConnectOptions::new()
                        .filename(&db_path)
                        .create_if_missing(false),
                )
                .await
                .unwrap();

            let (version,): (String,) =
                sqlx::query_as("SELECT value FROM metadata WHERE key = 'schema_version'")
                    .fetch_one(&pool)
                    .await
                    .unwrap();
            assert_eq!(
                version, "17",
                "schema_version must be restored to pre-migrate value 17"
            );

            let columns: Vec<(i64, String, String, i64, Option<String>, i64)> =
                sqlx::query_as("PRAGMA table_info(chunks)")
                    .fetch_all(&pool)
                    .await
                    .unwrap();
            assert!(
                !columns.iter().any(|(_, name, ..)| name == "embedding_base"),
                "the v17->v18 `embedding_base` column must be absent after restore"
            );
            pool.close().await;
        });
    }

    /// When a migration fails, the live pool must be closed BEFORE
    /// `restore_from_backup`'s `atomic_replace` runs over the DB file.
    /// Otherwise pool descriptors keep mmap'ing the unlinked old inode while
    /// subsequent opens see the restored backup — silent two-state
    /// divergence where in-process queries can read stale rows from the
    /// orphaned old inode while readers from new processes see the restored
    /// DB.
    ///
    /// This test simulates the daemon scenario: a long-lived
    /// pool is open, a migration runs and fails, then we verify that:
    /// 1. A fresh pool opened against the same path AFTER migrate returns
    ///    sees the restored pre-migrate state — proves the file replace
    ///    landed correctly.
    /// 2. The pool that was passed into migrate has been consumed (compile-
    ///    time enforcement via the value-taking signature). The error path
    ///    inside migrate closes the pool before atomic_replace, so the
    ///    descriptors against the orphaned inode are released.
    /// 3. The sentinel row written pre-migrate is readable through the
    ///    fresh pool. If migrate had skipped the close-and-restore protocol,
    ///    the file replace could have left the sentinel readable only
    ///    through the orphaned pool — which by now is gone.
    #[test]
    fn test_migrate_failure_closes_pool_before_restore_no_phantom_inode() {
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("test.db");

        // Phase 1: build a v17 DB with a sentinel row, checkpoint+close so
        // the on-disk bytes reflect the full state. The sentinel is what
        // we read back via the fresh pool after migration failure to
        // verify the restore landed on the inode the fresh open sees.
        rt.block_on(async {
            let pool = SqlitePoolOptions::new()
                .max_connections(1)
                .connect_with(
                    sqlx::sqlite::SqliteConnectOptions::new()
                        .filename(&db_path)
                        .create_if_missing(true)
                        .journal_mode(sqlx::sqlite::SqliteJournalMode::Wal)
                        .synchronous(sqlx::sqlite::SqliteSynchronous::Full),
                )
                .await
                .unwrap();
            sqlx::query("CREATE TABLE metadata (key TEXT PRIMARY KEY, value TEXT NOT NULL)")
                .execute(&pool)
                .await
                .unwrap();
            sqlx::query(
                "CREATE TABLE chunks (
                    id TEXT PRIMARY KEY,
                    origin TEXT NOT NULL,
                    source_type TEXT NOT NULL,
                    language TEXT NOT NULL,
                    chunk_type TEXT NOT NULL,
                    name TEXT NOT NULL,
                    signature TEXT NOT NULL,
                    content TEXT NOT NULL,
                    content_hash TEXT NOT NULL,
                    line_start INTEGER NOT NULL,
                    line_end INTEGER NOT NULL,
                    embedding BLOB NOT NULL,
                    created_at TEXT NOT NULL,
                    updated_at TEXT NOT NULL,
                    enrichment_version INTEGER NOT NULL DEFAULT 0
                )",
            )
            .execute(&pool)
            .await
            .unwrap();
            sqlx::query("INSERT INTO metadata (key, value) VALUES ('schema_version', '17')")
                .execute(&pool)
                .await
                .unwrap();
            // Sentinel row — the value is what we assert post-restore.
            sqlx::query(
                "INSERT INTO chunks (id, origin, source_type, language, chunk_type, \
                    name, signature, content, content_hash, line_start, line_end, \
                    embedding, created_at, updated_at) \
                 VALUES ('sentinel-1', 'file:lib.rs', 'file', 'rust', 'function', \
                    'sentinel_marker', 'fn sentinel()', 'fn sentinel() {}', \
                    'pre_migrate_hash', 1, 5, X'cafe', '2026-04-25', '2026-04-25')",
            )
            .execute(&pool)
            .await
            .unwrap();
            sqlx::query("PRAGMA wal_checkpoint(TRUNCATE)")
                .execute(&pool)
                .await
                .unwrap();
            pool.close().await;
        });

        // Phase 2: open a FRESH pool and run migrate(17 -> 19). The hook
        // makes migration step v18 fire its DDL then fail — the failure
        // path must close this pool internally before atomic_replace
        // restores the backup. We don't keep a handle to the pool after
        // migrate consumes it; the value-taking signature enforces that.
        TEST_FAIL_AFTER_VERSION.with(|c| c.set(18));

        rt.block_on(async {
            let pool = SqlitePoolOptions::new()
                .max_connections(1)
                .connect_with(
                    sqlx::sqlite::SqliteConnectOptions::new()
                        .filename(&db_path)
                        .create_if_missing(false)
                        .journal_mode(sqlx::sqlite::SqliteJournalMode::Wal)
                        .synchronous(sqlx::sqlite::SqliteSynchronous::Full),
                )
                .await
                .unwrap();

            let err = migrate(pool, &db_path, 17, 19).await.unwrap_err();
            // pool is consumed at this point — the borrow checker would
            // reject any further use. That alone enforces the close-before-
            // restore invariant: migrate cannot return Err with a still-open
            // pool because the signature returns SqlitePool only on Ok.
            assert!(
                matches!(&err, StoreError::Runtime(msg) if msg.contains("injected failure")),
                "expected injected-failure error, got: {:?}",
                err
            );
        });

        TEST_FAIL_AFTER_VERSION.with(|c| c.set(0));

        // Phase 3: open a FRESH pool against the same path. This simulates
        // a CLI invocation arriving after the daemon's failed migration —
        // it must see the restored DB on the inode the path resolves to,
        // not an orphaned post-DDL state.
        rt.block_on(async {
            let pool = SqlitePoolOptions::new()
                .max_connections(1)
                .connect_with(
                    sqlx::sqlite::SqliteConnectOptions::new()
                        .filename(&db_path)
                        .create_if_missing(false)
                        .journal_mode(sqlx::sqlite::SqliteJournalMode::Wal)
                        .synchronous(sqlx::sqlite::SqliteSynchronous::Full),
                )
                .await
                .unwrap();

            // Schema must be back at v17 — the v18 ALTER was rolled back
            // and the DB on disk is whatever the backup captured.
            let (version,): (String,) =
                sqlx::query_as("SELECT value FROM metadata WHERE key = 'schema_version'")
                    .fetch_one(&pool)
                    .await
                    .unwrap();
            assert_eq!(
                version, "17",
                "fresh pool must see pre-migrate schema_version on the restored inode"
            );

            // Sentinel row from phase 1 must be readable. If atomic_replace
            // had landed on a different inode than the fresh pool resolves,
            // either the row would be missing or the WAL/SHM mismatch
            // would yield a "database disk image is malformed" error.
            let row: Option<(String, String)> =
                sqlx::query_as("SELECT name, content_hash FROM chunks WHERE id = 'sentinel-1'")
                    .fetch_optional(&pool)
                    .await
                    .expect(
                        "fresh-pool query against the restored DB must succeed — \
                 if it errors, atomic_replace landed on a phantom inode",
                    );
            assert_eq!(
                row,
                Some((
                    "sentinel_marker".to_string(),
                    "pre_migrate_hash".to_string()
                )),
                "sentinel row from pre-migrate state must be readable via fresh pool — \
                 if it's missing or mutated, the file replace happened against an \
                 orphaned inode while the live pool was still mmap'd"
            );

            // The v18 column must NOT exist — confirms the failed v17→v18
            // ALTER did not leak through the restore.
            let columns: Vec<(String,)> =
                sqlx::query_as("SELECT name FROM pragma_table_info('chunks')")
                    .fetch_all(&pool)
                    .await
                    .unwrap();
            let has_embedding_base = columns.iter().any(|(n,)| n == "embedding_base");
            assert!(
                !has_embedding_base,
                "embedding_base (v18 column) must NOT exist after failed v17→v18 + restore — \
                 if present, the DDL leaked past the restore"
            );

            pool.close().await;
        });
    }

    /// Prune policy: after a successful migrate, only the newest
    /// `keep_backups()` (default 3) backups survive. Older ones are deleted.
    ///
    /// Setup seeds 5 fake `.bak-v*.db` files with staggered mtimes so there
    /// is a deterministic ordering, runs migrate which creates a 6th (the
    /// one produced by this run), and asserts that 3 survive: the two
    /// newest pre-existing backups and this run's backup.
    #[test]
    fn test_migrate_prunes_old_backups_keeping_newest_two() {
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("test.db");

        // Seed 5 pre-existing fake backups with distinct mtimes. Using
        // different filenames per from/to spans because the filename
        // already encodes a unique ts suffix in real usage; here we set
        // mtimes explicitly to control the sort order.
        let now = std::time::SystemTime::now();
        let five_secs = std::time::Duration::from_secs(5);
        let fake_backups = [
            ("test.bak-v10-v11-100.db", now - five_secs * 5), // oldest
            ("test.bak-v11-v12-200.db", now - five_secs * 4),
            ("test.bak-v12-v13-300.db", now - five_secs * 3),
            ("test.bak-v13-v14-400.db", now - five_secs * 2),
            ("test.bak-v14-v15-500.db", now - five_secs), // newest pre-existing
        ];
        for (name, mtime) in &fake_backups {
            let path = dir.path().join(name);
            // Valid SQLite header so any future "is this a real DB" check
            // wouldn't reject these synthetic ones during pruning.
            std::fs::write(&path, b"SQLite format 3\0filler").unwrap();
            // File::set_modified is stable as of Rust 1.75; cqs MSRV is 1.93.
            let f = std::fs::OpenOptions::new().write(true).open(&path).unwrap();
            f.set_modified(*mtime).unwrap();
        }

        rt.block_on(async {
            let pool = SqlitePoolOptions::new()
                .max_connections(1)
                .connect_with(
                    sqlx::sqlite::SqliteConnectOptions::new()
                        .filename(&db_path)
                        .create_if_missing(true)
                        .foreign_keys(true),
                )
                .await
                .unwrap();
            seed_v19_schema(&pool).await;

            let _pool = migrate(pool, &db_path, 19, 20).await.unwrap();
        });

        // After migrate, the prune keeps keep_backups() (default 3) newest. This run
        // just produced a new backup (newer mtime than any fake), so the
        // survivors must be: the two newest fakes (v13-v14, v14-v15) +
        // this run's v19-v20. That is three files total.
        let mut survivors: Vec<String> = std::fs::read_dir(dir.path())
            .unwrap()
            .flatten()
            .filter_map(|e| e.file_name().to_str().map(|s| s.to_string()))
            .filter(|n| n.starts_with("test.bak-v") && n.ends_with(".db"))
            .collect();
        survivors.sort();
        assert_eq!(
            survivors.len(),
            3,
            "expected 3 surviving backups (2 newest pre-existing + this run), got: {:?}",
            survivors
        );
        // Explicit: the oldest three pre-existing backups must have been pruned.
        for deleted in &[
            "test.bak-v10-v11-100.db",
            "test.bak-v11-v12-200.db",
            "test.bak-v12-v13-300.db",
        ] {
            assert!(
                !survivors.iter().any(|s| s == deleted),
                "{} should have been pruned",
                deleted
            );
        }
        // The two newest pre-existing backups must still be present.
        for kept in &["test.bak-v13-v14-400.db", "test.bak-v14-v15-500.db"] {
            assert!(
                survivors.iter().any(|s| s == kept),
                "{} should have been kept (among newest 2)",
                kept
            );
        }
    }

    /// Absent-WAL edge case: on a cleanly-closed SQLite DB the
    /// `-wal` and `-shm` sidecars can be absent. The backup must succeed
    /// without them (don't error on missing source), and the happy-path
    /// prune should leave a usable backup .db file behind.
    ///
    /// Regression for the shape where `copy_triplet` would bail on the
    /// first missing sidecar and leave the backup half-written.
    #[test]
    fn test_migrate_backup_works_when_wal_is_absent() {
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("test.db");

        // Phase 1: fully write the v19 schema, then TRUNCATE the WAL so
        // there are no sidecar files on disk before migrate() runs.
        rt.block_on(async {
            let pool = SqlitePoolOptions::new()
                .max_connections(1)
                .connect_with(
                    sqlx::sqlite::SqliteConnectOptions::new()
                        .filename(&db_path)
                        .create_if_missing(true)
                        .journal_mode(sqlx::sqlite::SqliteJournalMode::Wal)
                        .foreign_keys(true),
                )
                .await
                .unwrap();
            seed_v19_schema(&pool).await;
            sqlx::query("PRAGMA wal_checkpoint(TRUNCATE)")
                .execute(&pool)
                .await
                .unwrap();
            pool.close().await;
        });

        // Confirm the pre-migration state matches the scenario we want to
        // exercise: main DB present, no sidecars. (On a WAL-mode DB,
        // opening the pool below will recreate them.)
        assert!(db_path.exists(), "main DB should exist");
        let wal_sidecar = {
            let mut s = db_path.as_os_str().to_os_string();
            s.push("-wal");
            std::path::PathBuf::from(s)
        };
        assert!(
            !wal_sidecar.exists(),
            "precondition: no -wal sidecar on disk"
        );

        // Phase 2: reopen and migrate. Backup path should succeed.
        rt.block_on(async {
            let pool = SqlitePoolOptions::new()
                .max_connections(1)
                .connect_with(
                    sqlx::sqlite::SqliteConnectOptions::new()
                        .filename(&db_path)
                        .create_if_missing(false)
                        .journal_mode(sqlx::sqlite::SqliteJournalMode::Wal)
                        .foreign_keys(true),
                )
                .await
                .unwrap();
            let pool = migrate(pool, &db_path, 19, 20).await.unwrap();
            pool.close().await;
        });

        // A valid backup .db exists and parses as SQLite.
        let backups: Vec<_> = std::fs::read_dir(dir.path())
            .unwrap()
            .flatten()
            .filter_map(|e| e.file_name().to_str().map(|s| s.to_string()))
            .filter(|n| n.starts_with("test.bak-v19-v20-") && n.ends_with(".db"))
            .collect();
        assert_eq!(
            backups.len(),
            1,
            "expected one backup .db from this run, got: {:?}",
            backups
        );
        let bak_path = dir.path().join(&backups[0]);
        let bak_bytes = std::fs::read(&bak_path).unwrap();
        assert!(
            bak_bytes.starts_with(b"SQLite format 3\0"),
            "backup should be a valid SQLite database file"
        );
    }

    /// A DB without a `schema_version` metadata row must end up with one
    /// stamped after `run_migration_tx` completes. A plain
    /// `UPDATE metadata SET value = ?1 WHERE key = 'schema_version'` would
    /// affect zero rows in this case, so the next open would re-run all the
    /// migrations and crash on "duplicate column name" / "table already exists".
    #[test]
    fn test_migration_creates_schema_version_row_if_missing() {
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("test.db");

        rt.block_on(async {
            let pool = SqlitePoolOptions::new()
                .max_connections(1)
                .connect_with(
                    sqlx::sqlite::SqliteConnectOptions::new()
                        .filename(&db_path)
                        .create_if_missing(true),
                )
                .await
                .unwrap();

            // Build a minimal v10 schema WITHOUT inserting a schema_version row.
            // (We use v10 as the migration baseline because that's where the
            // DDL chain starts in this codebase.)
            sqlx::query(
                "CREATE TABLE IF NOT EXISTS metadata (
                    key TEXT PRIMARY KEY,
                    value TEXT NOT NULL
                )",
            )
            .execute(&pool)
            .await
            .unwrap();
            sqlx::query(
                "CREATE TABLE IF NOT EXISTS chunks (
                    id TEXT PRIMARY KEY,
                    origin TEXT NOT NULL,
                    language TEXT NOT NULL DEFAULT '',
                    chunk_type TEXT NOT NULL DEFAULT '',
                    name TEXT NOT NULL,
                    signature TEXT NOT NULL DEFAULT '',
                    content TEXT NOT NULL,
                    doc TEXT,
                    line_start INTEGER NOT NULL DEFAULT 0,
                    line_end INTEGER NOT NULL DEFAULT 0,
                    parent_id TEXT
                )",
            )
            .execute(&pool)
            .await
            .unwrap();

            // Verify the metadata row is absent before migration.
            let pre: Option<(String,)> =
                sqlx::query_as("SELECT value FROM metadata WHERE key = 'schema_version'")
                    .fetch_optional(&pool)
                    .await
                    .unwrap();
            assert!(
                pre.is_none(),
                "precondition: schema_version row must be absent"
            );

            // Run a single-step migration (10 -> 11). The caller's `from`
            // argument is what `run_migration_tx` falls back to when the
            // metadata row is missing.
            run_migration_tx(&pool, 10, 11).await.unwrap();

            // The row must now exist with the new version stamped on it.
            let post: Option<(String,)> =
                sqlx::query_as("SELECT value FROM metadata WHERE key = 'schema_version'")
                    .fetch_optional(&pool)
                    .await
                    .unwrap();
            assert_eq!(
                post.map(|(v,)| v),
                Some("11".to_string()),
                "schema_version row must be created with the new version"
            );
        });
    }

    /// v20→v21 adds a `parser_version` column that defaults to 0 for
    /// existing rows. New rows can write any u32 value;
    /// the UPSERT path uses `OR parser_version != excluded.parser_version`
    /// to refresh chunks whose source bytes are unchanged but whose parser
    /// emitted a different `doc` (or other non-content field).
    ///
    /// Round-trip:
    /// - Pre-migration v20 chunk gets `parser_version = 0` after ALTER.
    /// - Post-migration v21 INSERT can stamp any value (e.g. 5) and read it back.
    #[test]
    fn test_migrate_v20_to_v21_adds_parser_version_column() {
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("test.db");

        rt.block_on(async {
            let pool = SqlitePoolOptions::new()
                .max_connections(1)
                .connect_with(
                    sqlx::sqlite::SqliteConnectOptions::new()
                        .filename(&db_path)
                        .create_if_missing(true)
                        .foreign_keys(true),
                )
                .await
                .unwrap();

            // Minimal v20 schema: chunks with the v20 column set, metadata.
            // Notably no `parser_version` column — that's what v20→v21 adds.
            sqlx::query("CREATE TABLE metadata (key TEXT PRIMARY KEY, value TEXT NOT NULL)")
                .execute(&pool)
                .await
                .unwrap();
            sqlx::query(
                "CREATE TABLE chunks (
                    id TEXT PRIMARY KEY,
                    origin TEXT NOT NULL,
                    source_type TEXT NOT NULL,
                    language TEXT NOT NULL,
                    chunk_type TEXT NOT NULL,
                    name TEXT NOT NULL,
                    signature TEXT NOT NULL,
                    content TEXT NOT NULL,
                    content_hash TEXT NOT NULL,
                    line_start INTEGER NOT NULL,
                    line_end INTEGER NOT NULL,
                    embedding BLOB NOT NULL,
                    embedding_base BLOB,
                    created_at TEXT NOT NULL,
                    updated_at TEXT NOT NULL,
                    enrichment_version INTEGER NOT NULL DEFAULT 0
                )",
            )
            .execute(&pool)
            .await
            .unwrap();
            sqlx::query("INSERT INTO metadata (key, value) VALUES ('schema_version', '20')")
                .execute(&pool)
                .await
                .unwrap();

            // Seed a pre-migration v20 chunk.
            sqlx::query(
                "INSERT INTO chunks (id, origin, source_type, language, chunk_type, name, \
                 signature, content, content_hash, line_start, line_end, embedding, \
                 created_at, updated_at) \
                 VALUES ('pre_v21', 'file:lib.rs', 'file', 'rust', 'function', 'pre_v21', \
                 '', '', 'h', 1, 10, X'00', '2026-04-12', '2026-04-12')",
            )
            .execute(&pool)
            .await
            .unwrap();

            // Run v20 → v21 migration.
            let pool = migrate(pool, &db_path, 20, 21).await.unwrap();

            // Schema version bumped.
            let (v,): (String,) =
                sqlx::query_as("SELECT value FROM metadata WHERE key = 'schema_version'")
                    .fetch_one(&pool)
                    .await
                    .unwrap();
            assert_eq!(v, "21");

            // Pre-migration row has parser_version = 0 (default).
            let (pre_pv,): (i64,) =
                sqlx::query_as("SELECT parser_version FROM chunks WHERE id = 'pre_v21'")
                    .fetch_one(&pool)
                    .await
                    .unwrap();
            assert_eq!(
                pre_pv, 0,
                "v20 chunk after migration must default to parser_version = 0"
            );

            // New row written with parser_version = 5 round-trips correctly.
            sqlx::query(
                "INSERT INTO chunks (id, origin, source_type, language, chunk_type, name, \
                 signature, content, content_hash, line_start, line_end, embedding, \
                 created_at, updated_at, parser_version) \
                 VALUES ('post_v21', 'file:lib.rs', 'file', 'rust', 'function', 'post_v21', \
                 '', '', 'h', 1, 10, X'00', '2026-04-12', '2026-04-12', 5)",
            )
            .execute(&pool)
            .await
            .unwrap();
            let (post_pv,): (i64,) =
                sqlx::query_as("SELECT parser_version FROM chunks WHERE id = 'post_v21'")
                    .fetch_one(&pool)
                    .await
                    .unwrap();
            assert_eq!(
                post_pv, 5,
                "v21 INSERT with parser_version = 5 must round-trip"
            );
        });
    }

    /// v21 → v22: ALTER TABLE chunks ADD umap_x/umap_y REAL (both nullable).
    /// Round-trip: pre-migration v21 chunk gets NULL coords; post-migration
    /// v22 INSERT can stamp arbitrary floats and read them back. Negative,
    /// large-magnitude, and zero values all preserve round-trip.
    #[test]
    fn test_migrate_v21_to_v22_adds_umap_columns() {
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("test.db");

        rt.block_on(async {
            let pool = SqlitePoolOptions::new()
                .max_connections(1)
                .connect_with(
                    sqlx::sqlite::SqliteConnectOptions::new()
                        .filename(&db_path)
                        .create_if_missing(true)
                        .foreign_keys(true),
                )
                .await
                .unwrap();

            // Minimal v21 schema — same shape as v20 plus parser_version.
            // Notably no umap_x/umap_y columns — that's what v22 adds.
            sqlx::query("CREATE TABLE metadata (key TEXT PRIMARY KEY, value TEXT NOT NULL)")
                .execute(&pool)
                .await
                .unwrap();
            sqlx::query(
                "CREATE TABLE chunks (
                    id TEXT PRIMARY KEY,
                    origin TEXT NOT NULL,
                    source_type TEXT NOT NULL,
                    language TEXT NOT NULL,
                    chunk_type TEXT NOT NULL,
                    name TEXT NOT NULL,
                    signature TEXT NOT NULL,
                    content TEXT NOT NULL,
                    content_hash TEXT NOT NULL,
                    line_start INTEGER NOT NULL,
                    line_end INTEGER NOT NULL,
                    embedding BLOB NOT NULL,
                    embedding_base BLOB,
                    created_at TEXT NOT NULL,
                    updated_at TEXT NOT NULL,
                    enrichment_version INTEGER NOT NULL DEFAULT 0,
                    parser_version INTEGER NOT NULL DEFAULT 0
                )",
            )
            .execute(&pool)
            .await
            .unwrap();
            sqlx::query("INSERT INTO metadata (key, value) VALUES ('schema_version', '21')")
                .execute(&pool)
                .await
                .unwrap();

            // Seed a pre-migration v21 chunk.
            sqlx::query(
                "INSERT INTO chunks (id, origin, source_type, language, chunk_type, name, \
                 signature, content, content_hash, line_start, line_end, embedding, \
                 created_at, updated_at) \
                 VALUES ('pre_v22', 'file:lib.rs', 'file', 'rust', 'function', 'pre_v22', \
                 '', '', 'h', 1, 10, X'00', '2026-04-21', '2026-04-21')",
            )
            .execute(&pool)
            .await
            .unwrap();

            // Run v21 → v22 migration.
            let pool = migrate(pool, &db_path, 21, 22).await.unwrap();

            // Schema version bumped.
            let (v,): (String,) =
                sqlx::query_as("SELECT value FROM metadata WHERE key = 'schema_version'")
                    .fetch_one(&pool)
                    .await
                    .unwrap();
            assert_eq!(v, "22");

            // Pre-migration row has NULL coords (no DEFAULT — UMAP is opt-in).
            let (pre_x, pre_y): (Option<f64>, Option<f64>) =
                sqlx::query_as("SELECT umap_x, umap_y FROM chunks WHERE id = 'pre_v22'")
                    .fetch_one(&pool)
                    .await
                    .unwrap();
            assert!(pre_x.is_none(), "pre-migration umap_x must be NULL");
            assert!(pre_y.is_none(), "pre-migration umap_y must be NULL");

            // Round-trip a few representative values: positive, negative, zero.
            for (id, x, y) in [
                ("post_v22_a", 1.5_f64, -2.25_f64),
                ("post_v22_b", 0.0_f64, 0.0_f64),
                ("post_v22_c", 1234.567_f64, -9876.54321_f64),
            ] {
                sqlx::query(
                    "INSERT INTO chunks (id, origin, source_type, language, chunk_type, name, \
                     signature, content, content_hash, line_start, line_end, embedding, \
                     created_at, updated_at, umap_x, umap_y) \
                     VALUES (?, 'file:lib.rs', 'file', 'rust', 'function', 'post_v22', \
                     '', '', 'h', 1, 10, X'00', '2026-04-21', '2026-04-21', ?, ?)",
                )
                .bind(id)
                .bind(x)
                .bind(y)
                .execute(&pool)
                .await
                .unwrap();
                let (rx, ry): (f64, f64) =
                    sqlx::query_as("SELECT umap_x, umap_y FROM chunks WHERE id = ?")
                        .bind(id)
                        .fetch_one(&pool)
                        .await
                        .unwrap();
                assert_eq!(rx, x, "umap_x round-trip for {id}");
                assert_eq!(ry, y, "umap_y round-trip for {id}");
            }
        });
    }

    /// v22 → v23: ALTER TABLE chunks ADD source_size INTEGER + source_content_hash BLOB
    /// (both nullable). Round-trip: pre-migration v22 chunk gets NULL fingerprint;
    /// post-migration v23 INSERT can stamp arbitrary size + 32-byte hash and read
    /// them back. Reconcile fingerprint can decide divergence using mtime, size, or
    /// content_hash interchangeably.
    #[test]
    fn test_migrate_v22_to_v23_adds_fingerprint_columns() {
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("test.db");

        rt.block_on(async {
            let pool = SqlitePoolOptions::new()
                .max_connections(1)
                .connect_with(
                    sqlx::sqlite::SqliteConnectOptions::new()
                        .filename(&db_path)
                        .create_if_missing(true)
                        .foreign_keys(true),
                )
                .await
                .unwrap();

            // Minimal v22 schema — same shape as v21 plus umap_x/umap_y.
            // Notably no source_size/source_content_hash columns — that's what v23 adds.
            sqlx::query("CREATE TABLE metadata (key TEXT PRIMARY KEY, value TEXT NOT NULL)")
                .execute(&pool)
                .await
                .unwrap();
            sqlx::query(
                "CREATE TABLE chunks (
                    id TEXT PRIMARY KEY,
                    origin TEXT NOT NULL,
                    source_type TEXT NOT NULL,
                    language TEXT NOT NULL,
                    chunk_type TEXT NOT NULL,
                    name TEXT NOT NULL,
                    signature TEXT NOT NULL,
                    content TEXT NOT NULL,
                    content_hash TEXT NOT NULL,
                    line_start INTEGER NOT NULL,
                    line_end INTEGER NOT NULL,
                    embedding BLOB NOT NULL,
                    embedding_base BLOB,
                    created_at TEXT NOT NULL,
                    updated_at TEXT NOT NULL,
                    enrichment_version INTEGER NOT NULL DEFAULT 0,
                    parser_version INTEGER NOT NULL DEFAULT 0,
                    umap_x REAL,
                    umap_y REAL
                )",
            )
            .execute(&pool)
            .await
            .unwrap();
            sqlx::query("INSERT INTO metadata (key, value) VALUES ('schema_version', '22')")
                .execute(&pool)
                .await
                .unwrap();

            // Seed a pre-migration v22 chunk.
            sqlx::query(
                "INSERT INTO chunks (id, origin, source_type, language, chunk_type, name, \
                 signature, content, content_hash, line_start, line_end, embedding, \
                 created_at, updated_at) \
                 VALUES ('pre_v23', 'file:lib.rs', 'file', 'rust', 'function', 'pre_v23', \
                 '', '', 'h', 1, 10, X'00', '2026-04-29', '2026-04-29')",
            )
            .execute(&pool)
            .await
            .unwrap();

            // Run v22 → v23 migration.
            let pool = migrate(pool, &db_path, 22, 23).await.unwrap();

            // Schema version bumped.
            let (v,): (String,) =
                sqlx::query_as("SELECT value FROM metadata WHERE key = 'schema_version'")
                    .fetch_one(&pool)
                    .await
                    .unwrap();
            assert_eq!(v, "23");

            // Pre-migration row has NULL fingerprint (no DEFAULT — populated on
            // first re-embed of the file).
            let (pre_size, pre_hash): (Option<i64>, Option<Vec<u8>>) = sqlx::query_as(
                "SELECT source_size, source_content_hash FROM chunks WHERE id = 'pre_v23'",
            )
            .fetch_one(&pool)
            .await
            .unwrap();
            assert!(pre_size.is_none(), "pre-migration source_size must be NULL");
            assert!(
                pre_hash.is_none(),
                "pre-migration source_content_hash must be NULL"
            );

            // Round-trip representative fingerprints: zero size, large size, all-zero
            // hash, BLAKE3-shaped 32-byte hash.
            let cases: &[(&str, i64, Vec<u8>)] = &[
                ("post_v23_empty", 0_i64, vec![0u8; 32]),
                ("post_v23_typical", 4096_i64, (0u8..32).collect::<Vec<u8>>()),
                ("post_v23_huge", 10_485_760_i64, vec![0xffu8; 32]),
            ];
            for (id, size, hash) in cases {
                sqlx::query(
                    "INSERT INTO chunks (id, origin, source_type, language, chunk_type, name, \
                     signature, content, content_hash, line_start, line_end, embedding, \
                     created_at, updated_at, source_size, source_content_hash) \
                     VALUES (?, 'file:lib.rs', 'file', 'rust', 'function', 'post_v23', \
                     '', '', 'h', 1, 10, X'00', '2026-04-29', '2026-04-29', ?, ?)",
                )
                .bind(id)
                .bind(*size)
                .bind(hash)
                .execute(&pool)
                .await
                .unwrap();
                let (rsize, rhash): (i64, Vec<u8>) = sqlx::query_as(
                    "SELECT source_size, source_content_hash FROM chunks WHERE id = ?",
                )
                .bind(id)
                .fetch_one(&pool)
                .await
                .unwrap();
                assert_eq!(rsize, *size, "source_size round-trip for {id}");
                assert_eq!(&rhash, hash, "source_content_hash round-trip for {id}");
            }
        });
    }

    /// v27 / #1452: `chunks.needs_embedding` column added with DEFAULT 0.
    /// Pre-migration rows must default to 0 (preserves "all chunks are
    /// embedded" invariant from v26 era). The partial index
    /// `idx_chunks_needs_embedding WHERE needs_embedding=1` is created so
    /// `needs_embedding_count` is O(needs) regardless of total rows.
    #[test]
    fn test_migrate_v26_to_v27_adds_needs_embedding_flag() {
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("test.db");

        rt.block_on(async {
            let pool = SqlitePoolOptions::new()
                .max_connections(1)
                .connect_with(
                    sqlx::sqlite::SqliteConnectOptions::new()
                        .filename(&db_path)
                        .create_if_missing(true)
                        .foreign_keys(true),
                )
                .await
                .unwrap();

            // Minimal v26 schema — pre-needs_embedding shape.
            sqlx::query("CREATE TABLE metadata (key TEXT PRIMARY KEY, value TEXT NOT NULL)")
                .execute(&pool)
                .await
                .unwrap();
            sqlx::query(
                "CREATE TABLE chunks (
                    id TEXT PRIMARY KEY,
                    origin TEXT NOT NULL,
                    source_type TEXT NOT NULL,
                    language TEXT NOT NULL,
                    chunk_type TEXT NOT NULL,
                    name TEXT NOT NULL,
                    signature TEXT NOT NULL,
                    content TEXT NOT NULL,
                    content_hash TEXT NOT NULL,
                    line_start INTEGER NOT NULL,
                    line_end INTEGER NOT NULL,
                    embedding BLOB NOT NULL,
                    embedding_base BLOB,
                    created_at TEXT NOT NULL,
                    updated_at TEXT NOT NULL,
                    enrichment_version INTEGER NOT NULL DEFAULT 0,
                    parser_version INTEGER NOT NULL DEFAULT 0,
                    vendored INTEGER NOT NULL DEFAULT 0
                )",
            )
            .execute(&pool)
            .await
            .unwrap();
            sqlx::query("INSERT INTO metadata (key, value) VALUES ('schema_version', '26')")
                .execute(&pool)
                .await
                .unwrap();

            // Seed a pre-migration v26 chunk WITH embedding_base populated
            // (the post-v17 invariant for properly-indexed rows).
            sqlx::query(
                "INSERT INTO chunks (id, origin, source_type, language, chunk_type, name, \
                 signature, content, content_hash, line_start, line_end, embedding, \
                 embedding_base, created_at, updated_at) \
                 VALUES ('pre_v27', 'file:lib.rs', 'file', 'rust', 'function', 'pre_v27', \
                 '', '', 'h', 1, 10, X'00', X'00', '2026-05-06', '2026-05-06')",
            )
            .execute(&pool)
            .await
            .unwrap();

            // Seed a row with NULL embedding_base — represents chunks where
            // the base column was never filled. The migration must mark
            // these as needs_embedding=1 so the next index pass triggers a
            // re-embed and the base-HNSW coverage gap closes.
            sqlx::query(
                "INSERT INTO chunks (id, origin, source_type, language, chunk_type, name, \
                 signature, content, content_hash, line_start, line_end, embedding, \
                 created_at, updated_at) \
                 VALUES ('pre_v27_no_base', 'file:lib.rs', 'file', 'rust', 'function', \
                 'pre_v27_no_base', '', '', 'h_nb', 1, 10, X'00', '2026-05-06', '2026-05-06')",
            )
            .execute(&pool)
            .await
            .unwrap();

            // Run v26 → v27 migration.
            let pool = migrate(pool, &db_path, 26, 27).await.unwrap();

            // Schema version bumped.
            let (v,): (String,) =
                sqlx::query_as("SELECT value FROM metadata WHERE key = 'schema_version'")
                    .fetch_one(&pool)
                    .await
                    .unwrap();
            assert_eq!(v, "27");

            // Properly-indexed pre-migration row has needs_embedding = 0
            // (DEFAULT). Existing chunks with a real `embedding_base` were
            // embedded under the v26 contract; the migration must not flip
            // them to "needs embedding" or every reindex would re-enrich
            // them and waste GPU.
            let (pre_flag,): (i64,) =
                sqlx::query_as("SELECT needs_embedding FROM chunks WHERE id = 'pre_v27'")
                    .fetch_one(&pool)
                    .await
                    .unwrap();
            assert_eq!(
                pre_flag, 0,
                "pre-migration chunk with real embedding_base must retain needs_embedding=0"
            );

            // Rows with NULL embedding_base get backfilled to
            // needs_embedding=1 so the next index pass repopulates the
            // base column.
            let (no_base_flag,): (i64,) =
                sqlx::query_as("SELECT needs_embedding FROM chunks WHERE id = 'pre_v27_no_base'")
                    .fetch_one(&pool)
                    .await
                    .unwrap();
            assert_eq!(
                no_base_flag, 1,
                "DS-V1.38-8: pre-migration chunk with NULL embedding_base must be marked \
                 needs_embedding=1 so the next enrichment_pass repopulates the base column"
            );

            // The partial index exists.
            let (idx_count,): (i64,) = sqlx::query_as(
                "SELECT COUNT(*) FROM sqlite_master \
                 WHERE type = 'index' AND name = 'idx_chunks_needs_embedding'",
            )
            .fetch_one(&pool)
            .await
            .unwrap();
            assert_eq!(
                idx_count, 1,
                "v27 must create idx_chunks_needs_embedding partial index"
            );

            // Insert a fresh chunk with needs_embedding=1 and round-trip it.
            sqlx::query(
                "INSERT INTO chunks (id, origin, source_type, language, chunk_type, name, \
                 signature, content, content_hash, line_start, line_end, embedding, \
                 created_at, updated_at, needs_embedding) \
                 VALUES ('post_v27_unembedded', 'file:lib.rs', 'file', 'rust', 'function', \
                 'post_v27_unembedded', '', '', 'h2', 1, 10, X'00', '2026-05-06', '2026-05-06', 1)",
            )
            .execute(&pool)
            .await
            .unwrap();
            let (round_flag,): (i64,) = sqlx::query_as(
                "SELECT needs_embedding FROM chunks WHERE id = 'post_v27_unembedded'",
            )
            .fetch_one(&pool)
            .await
            .unwrap();
            assert_eq!(round_flag, 1);
        });
    }

    /// v27 → v28 adds `chunks.canonical_hash` (nullable) + the
    /// `idx_chunks_canonical_hash` index. Pre-v28 rows keep `canonical_hash`
    /// NULL (clean cache miss); fresh rows can write it. Verifies the column,
    /// the index, NULL on the pre-migration row, and a round-trip write.
    #[test]
    fn test_migrate_v27_to_v28_adds_canonical_hash_column() {
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("test.db");

        rt.block_on(async {
            let pool = SqlitePoolOptions::new()
                .max_connections(1)
                .connect_with(
                    sqlx::sqlite::SqliteConnectOptions::new()
                        .filename(&db_path)
                        .create_if_missing(true)
                        .foreign_keys(true),
                )
                .await
                .unwrap();

            // Minimal v27 schema — chunks WITHOUT canonical_hash.
            sqlx::query("CREATE TABLE metadata (key TEXT PRIMARY KEY, value TEXT NOT NULL)")
                .execute(&pool)
                .await
                .unwrap();
            sqlx::query(
                "CREATE TABLE chunks (
                    id TEXT PRIMARY KEY,
                    origin TEXT NOT NULL,
                    source_type TEXT NOT NULL,
                    language TEXT NOT NULL,
                    chunk_type TEXT NOT NULL,
                    name TEXT NOT NULL,
                    signature TEXT NOT NULL,
                    content TEXT NOT NULL,
                    content_hash TEXT NOT NULL,
                    line_start INTEGER NOT NULL,
                    line_end INTEGER NOT NULL,
                    embedding BLOB NOT NULL,
                    embedding_base BLOB,
                    created_at TEXT NOT NULL,
                    updated_at TEXT NOT NULL,
                    enrichment_version INTEGER NOT NULL DEFAULT 0,
                    parser_version INTEGER NOT NULL DEFAULT 0,
                    vendored INTEGER NOT NULL DEFAULT 0,
                    needs_embedding INTEGER NOT NULL DEFAULT 0
                )",
            )
            .execute(&pool)
            .await
            .unwrap();
            sqlx::query("INSERT INTO metadata (key, value) VALUES ('schema_version', '27')")
                .execute(&pool)
                .await
                .unwrap();

            // Seed a pre-migration v27 chunk.
            sqlx::query(
                "INSERT INTO chunks (id, origin, source_type, language, chunk_type, name, \
                 signature, content, content_hash, line_start, line_end, embedding, \
                 created_at, updated_at) \
                 VALUES ('pre_v28', 'file:lib.rs', 'file', 'rust', 'function', 'pre_v28', \
                 '', '', 'h', 1, 10, X'00', '2026-06-09', '2026-06-09')",
            )
            .execute(&pool)
            .await
            .unwrap();

            // Run v27 → v28 migration.
            let pool = migrate(pool, &db_path, 27, 28).await.unwrap();

            // Schema version bumped.
            let (v,): (String,) =
                sqlx::query_as("SELECT value FROM metadata WHERE key = 'schema_version'")
                    .fetch_one(&pool)
                    .await
                    .unwrap();
            assert_eq!(v, "28");

            // The column exists and the pre-migration row is NULL.
            let (canon,): (Option<String>,) =
                sqlx::query_as("SELECT canonical_hash FROM chunks WHERE id = 'pre_v28'")
                    .fetch_one(&pool)
                    .await
                    .unwrap();
            assert!(
                canon.is_none(),
                "pre-v28 row must have NULL canonical_hash (clean cache miss)"
            );

            // The index exists.
            let (idx_count,): (i64,) = sqlx::query_as(
                "SELECT COUNT(*) FROM sqlite_master \
                 WHERE type = 'index' AND name = 'idx_chunks_canonical_hash'",
            )
            .fetch_one(&pool)
            .await
            .unwrap();
            assert_eq!(
                idx_count, 1,
                "v28 must create idx_chunks_canonical_hash index"
            );

            // A fresh row can carry a non-NULL canonical_hash and round-trip.
            sqlx::query(
                "INSERT INTO chunks (id, origin, source_type, language, chunk_type, name, \
                 signature, content, content_hash, line_start, line_end, embedding, \
                 created_at, updated_at, canonical_hash) \
                 VALUES ('post_v28', 'file:lib.rs', 'file', 'rust', 'function', 'post_v28', \
                 '', '', 'h2', 1, 10, X'00', '2026-06-09', '2026-06-09', 'canon123')",
            )
            .execute(&pool)
            .await
            .unwrap();
            let (round,): (Option<String>,) =
                sqlx::query_as("SELECT canonical_hash FROM chunks WHERE id = 'post_v28'")
                    .fetch_one(&pool)
                    .await
                    .unwrap();
            assert_eq!(round.as_deref(), Some("canon123"));
        });
    }

    /// Build a minimal v28 schema (metadata + a notes table WITHOUT the
    /// sentiment CHECK, stamped at version 28) at `db_path`. Returns the open
    /// pool. Shared by the v28→v29 tests.
    async fn setup_v28_schema(db_path: &std::path::Path) -> SqlitePool {
        let pool = SqlitePoolOptions::new()
            .max_connections(1)
            .connect_with(
                sqlx::sqlite::SqliteConnectOptions::new()
                    .filename(db_path)
                    .create_if_missing(true)
                    .foreign_keys(true),
            )
            .await
            .unwrap();
        sqlx::query("CREATE TABLE metadata (key TEXT PRIMARY KEY, value TEXT NOT NULL)")
            .execute(&pool)
            .await
            .unwrap();
        // Pre-v29 notes table: sentiment is a bare REAL with no CHECK.
        sqlx::query(
            "CREATE TABLE notes (
                id TEXT PRIMARY KEY,
                text TEXT NOT NULL,
                sentiment REAL NOT NULL,
                mentions TEXT,
                embedding BLOB NOT NULL,
                source_file TEXT NOT NULL,
                file_mtime INTEGER NOT NULL,
                created_at TEXT NOT NULL,
                updated_at TEXT NOT NULL,
                kind TEXT
            )",
        )
        .execute(&pool)
        .await
        .unwrap();
        sqlx::query("INSERT INTO metadata (key, value) VALUES ('schema_version', '28')")
            .execute(&pool)
            .await
            .unwrap();
        pool
    }

    /// v28 → v29 creates the `file_registry` table #1774 that persists the
    /// reconcile fingerprint independent of chunk rows. Verifies the table
    /// exists, has the right columns, and round-trips an UPSERT.
    #[test]
    fn test_migrate_v28_to_v29_creates_file_registry() {
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("test.db");

        rt.block_on(async {
            let pool = setup_v28_schema(&db_path).await;
            let pool = migrate(pool, &db_path, 28, 29).await.unwrap();

            let (v,): (String,) =
                sqlx::query_as("SELECT value FROM metadata WHERE key = 'schema_version'")
                    .fetch_one(&pool)
                    .await
                    .unwrap();
            assert_eq!(v, "29");

            // The table exists.
            let (tbl_count,): (i64,) = sqlx::query_as(
                "SELECT COUNT(*) FROM sqlite_master \
                 WHERE type = 'table' AND name = 'file_registry'",
            )
            .fetch_one(&pool)
            .await
            .unwrap();
            assert_eq!(tbl_count, 1, "v29 must create the file_registry table");

            // Round-trip an UPSERT: insert, then conflict-update.
            sqlx::query(
                "INSERT INTO file_registry (origin, source_mtime, source_size, source_content_hash) \
                 VALUES ('src/empty.rs', 1000, 42, X'aabb')",
            )
            .execute(&pool)
            .await
            .unwrap();
            sqlx::query(
                "INSERT INTO file_registry (origin, source_mtime, source_size, source_content_hash) \
                 VALUES ('src/empty.rs', 2000, 84, X'ccdd') \
                 ON CONFLICT(origin) DO UPDATE SET \
                     source_mtime = excluded.source_mtime, \
                     source_size = excluded.source_size, \
                     source_content_hash = excluded.source_content_hash",
            )
            .execute(&pool)
            .await
            .unwrap();
            let (mtime, size): (Option<i64>, Option<i64>) = sqlx::query_as(
                "SELECT source_mtime, source_size FROM file_registry WHERE origin = 'src/empty.rs'",
            )
            .fetch_one(&pool)
            .await
            .unwrap();
            assert_eq!(mtime, Some(2000), "UPSERT must replace mtime");
            assert_eq!(size, Some(84), "UPSERT must replace size");

            // PRIMARY KEY on origin: exactly one row after the conflict.
            let (cnt,): (i64,) = sqlx::query_as("SELECT COUNT(*) FROM file_registry")
                .fetch_one(&pool)
                .await
                .unwrap();
            assert_eq!(cnt, 1, "origin is the PRIMARY KEY — no duplicate row");
        });
    }

    /// v28 → v29 adds the notes.sentiment CHECK and clamp-rewrites any
    /// pre-existing off-grid rows to the nearest discrete value (-1, -0.5, 0,
    /// 0.5, 1). Verifies the clamp, the CHECK rejection of a new off-grid
    /// insert, and that legal values still insert.
    #[test]
    fn test_migrate_v28_to_v29_sentiment_check_and_clamp() {
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("test.db");

        rt.block_on(async {
            let pool = setup_v28_schema(&db_path).await;

            // Seed off-grid notes the migration must clamp. -0.8 → -1.0,
            // 0.9 → 1.0, 0.2 → 0.0, -0.3 → -0.5, 5.0 (out of range) → 1.0.
            let seeds = [
                ("n1", -0.8_f64, -1.0_f64),
                ("n2", 0.9, 1.0),
                ("n3", 0.2, 0.0),
                ("n4", -0.3, -0.5),
                ("n5", 5.0, 1.0),
                ("n6", 0.5, 0.5), // already legal — must survive unchanged
            ];
            for (id, raw, _) in &seeds {
                sqlx::query(
                    "INSERT INTO notes (id, text, sentiment, mentions, embedding, \
                     source_file, file_mtime, created_at, updated_at) \
                     VALUES (?1, 'note', ?2, '[]', X'', 'notes.toml', 0, 'now', 'now')",
                )
                .bind(id)
                .bind(raw)
                .execute(&pool)
                .await
                .unwrap();
            }

            let pool = migrate(pool, &db_path, 28, 29).await.unwrap();

            let (v,): (String,) =
                sqlx::query_as("SELECT value FROM metadata WHERE key = 'schema_version'")
                    .fetch_one(&pool)
                    .await
                    .unwrap();
            assert_eq!(v, "29");

            // Every seeded row clamped to its expected discrete value.
            for (id, _, expected) in &seeds {
                let (got,): (f64,) = sqlx::query_as("SELECT sentiment FROM notes WHERE id = ?1")
                    .bind(id)
                    .fetch_one(&pool)
                    .await
                    .unwrap();
                assert_eq!(
                    got, *expected,
                    "note {id} sentiment must clamp to the nearest discrete value"
                );
            }

            // The CHECK now rejects an off-grid insert.
            let bad = sqlx::query(
                "INSERT INTO notes (id, text, sentiment, mentions, embedding, \
                 source_file, file_mtime, created_at, updated_at) \
                 VALUES ('bad', 'x', -0.8, '[]', X'', 'notes.toml', 0, 'now', 'now')",
            )
            .execute(&pool)
            .await;
            assert!(
                bad.is_err(),
                "post-v29 the CHECK must reject an off-grid sentiment"
            );

            // A legal value still inserts.
            sqlx::query(
                "INSERT INTO notes (id, text, sentiment, mentions, embedding, \
                 source_file, file_mtime, created_at, updated_at) \
                 VALUES ('good', 'x', -0.5, '[]', X'', 'notes.toml', 0, 'now', 'now')",
            )
            .execute(&pool)
            .await
            .expect("a legal discrete sentiment must insert under the v29 CHECK");

            // The notes indexes survived the table rebuild.
            let (idx_count,): (i64,) = sqlx::query_as(
                "SELECT COUNT(*) FROM sqlite_master \
                 WHERE type = 'index' AND name IN ('idx_notes_sentiment', 'idx_notes_kind')",
            )
            .fetch_one(&pool)
            .await
            .unwrap();
            assert_eq!(
                idx_count, 2,
                "v29 must recreate the notes indexes after rebuild"
            );
        });
    }

    /// Fresh-create (schema.sql) and migrated (v28 → v29) must produce the same
    /// shape for the two objects v29 touches: `file_registry` exists in both,
    /// and `notes` carries the sentiment CHECK in both. Compares the
    /// `sqlite_master.sql` text for the notes table CHECK token and the
    /// presence of the registry table.
    #[test]
    fn test_v29_fresh_create_matches_migrated_shape() {
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();

        rt.block_on(async {
            // Fresh-create: apply schema.sql to an in-memory DB.
            let fresh = SqlitePoolOptions::new()
                .max_connections(1)
                .connect_with(
                    sqlx::sqlite::SqliteConnectOptions::new()
                        .filename(":memory:")
                        .foreign_keys(true),
                )
                .await
                .unwrap();
            // Use the same BEGIN/END-aware splitter `Store::init` uses — a naive
            // split on `;` breaks the `bump_splade_on_chunks_delete` trigger
            // body (which contains internal `;`).
            for stmt in crate::store::split_sql_statements(include_str!("../schema.sql")) {
                if stmt.is_empty() {
                    continue;
                }
                sqlx::query(sqlx::AssertSqlSafe(stmt.as_str()))
                    .execute(&fresh)
                    .await
                    .unwrap();
            }

            // file_registry exists in the fresh schema.
            let (fresh_reg,): (i64,) = sqlx::query_as(
                "SELECT COUNT(*) FROM sqlite_master \
                 WHERE type = 'table' AND name = 'file_registry'",
            )
            .fetch_one(&fresh)
            .await
            .unwrap();
            assert_eq!(fresh_reg, 1, "fresh schema.sql must define file_registry");

            // Fresh notes table carries the sentiment CHECK.
            let (fresh_notes_sql,): (String,) = sqlx::query_as(
                "SELECT sql FROM sqlite_master WHERE type = 'table' AND name = 'notes'",
            )
            .fetch_one(&fresh)
            .await
            .unwrap();
            assert!(
                fresh_notes_sql.to_lowercase().contains("check"),
                "fresh notes table must carry the sentiment CHECK, got: {fresh_notes_sql}"
            );

            // Migrated: build v28, migrate to v29.
            let dir = tempfile::tempdir().unwrap();
            let db_path = dir.path().join("test.db");
            let pool = setup_v28_schema(&db_path).await;
            let pool = migrate(pool, &db_path, 28, 29).await.unwrap();

            let (mig_reg,): (i64,) = sqlx::query_as(
                "SELECT COUNT(*) FROM sqlite_master \
                 WHERE type = 'table' AND name = 'file_registry'",
            )
            .fetch_one(&pool)
            .await
            .unwrap();
            assert_eq!(mig_reg, 1, "migrated DB must define file_registry");

            let (mig_notes_sql,): (String,) = sqlx::query_as(
                "SELECT sql FROM sqlite_master WHERE type = 'table' AND name = 'notes'",
            )
            .fetch_one(&pool)
            .await
            .unwrap();
            assert!(
                mig_notes_sql.to_lowercase().contains("check"),
                "migrated notes table must carry the sentiment CHECK, got: {mig_notes_sql}"
            );
        });
    }

    /// Build a minimal v29 schema (metadata + a v29-shaped `function_calls`
    /// table WITHOUT the `edge_kind` column, stamped at version 29) at
    /// `db_path`. Returns the open pool. Shared by the v29→v30 tests.
    async fn setup_v29_schema(db_path: &std::path::Path) -> SqlitePool {
        let pool = SqlitePoolOptions::new()
            .max_connections(1)
            .connect_with(
                sqlx::sqlite::SqliteConnectOptions::new()
                    .filename(db_path)
                    .create_if_missing(true)
                    .foreign_keys(true),
            )
            .await
            .unwrap();
        sqlx::query("CREATE TABLE metadata (key TEXT PRIMARY KEY, value TEXT NOT NULL)")
            .execute(&pool)
            .await
            .unwrap();
        // Pre-v30 function_calls: no edge_kind column.
        sqlx::query(
            "CREATE TABLE function_calls (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                file TEXT NOT NULL,
                caller_name TEXT NOT NULL,
                caller_line INTEGER NOT NULL,
                callee_name TEXT NOT NULL,
                call_line INTEGER NOT NULL
            )",
        )
        .execute(&pool)
        .await
        .unwrap();
        sqlx::query("INSERT INTO metadata (key, value) VALUES ('schema_version', '29')")
            .execute(&pool)
            .await
            .unwrap();
        pool
    }

    /// v29 → v30 adds the `function_calls.edge_kind` column with a `'call'`
    /// default. A pre-existing row (inserted before the migration without the
    /// column) must read back `edge_kind = 'call'`, and a fresh insert can
    /// carry a non-default kind.
    #[test]
    fn test_migrate_v29_to_v30_adds_edge_kind_column() {
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("test.db");

        rt.block_on(async {
            let pool = setup_v29_schema(&db_path).await;

            // Insert a pre-migration row (no edge_kind column yet).
            sqlx::query(
                "INSERT INTO function_calls (file, caller_name, caller_line, callee_name, call_line) \
                 VALUES ('src/a.rs', 'caller', 1, 'callee', 2)",
            )
            .execute(&pool)
            .await
            .unwrap();

            let pool = migrate(pool, &db_path, 29, 30).await.unwrap();

            // Pre-migration row defaults to 'call'.
            let (kind,): (String,) =
                sqlx::query_as("SELECT edge_kind FROM function_calls WHERE callee_name = 'callee'")
                    .fetch_one(&pool)
                    .await
                    .unwrap();
            assert_eq!(
                kind, "call",
                "pre-v30 function_calls row must default to edge_kind = 'call'"
            );

            // A fresh insert can carry a heuristic kind.
            sqlx::query(
                "INSERT INTO function_calls (file, caller_name, caller_line, callee_name, call_line, edge_kind) \
                 VALUES ('src/b.rs', 'c2', 1, 'cb', 2, 'serde_callback')",
            )
            .execute(&pool)
            .await
            .unwrap();
            let (kind2,): (String,) =
                sqlx::query_as("SELECT edge_kind FROM function_calls WHERE callee_name = 'cb'")
                    .fetch_one(&pool)
                    .await
                    .unwrap();
            assert_eq!(kind2, "serde_callback");
        });
    }

    /// Fresh-create (schema.sql) and migrated (v29 → v30) must produce the same
    /// `function_calls` shape: both carry an `edge_kind` column defaulting to
    /// `'call'`. Compares the `sqlite_master.sql` text for the column presence.
    #[test]
    fn test_v30_fresh_create_matches_migrated_shape() {
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();

        rt.block_on(async {
            // Fresh-create: apply schema.sql to an in-memory DB.
            let fresh = SqlitePoolOptions::new()
                .max_connections(1)
                .connect_with(
                    sqlx::sqlite::SqliteConnectOptions::new()
                        .filename(":memory:")
                        .foreign_keys(true),
                )
                .await
                .unwrap();
            for stmt in crate::store::split_sql_statements(include_str!("../schema.sql")) {
                if stmt.is_empty() {
                    continue;
                }
                sqlx::query(sqlx::AssertSqlSafe(stmt.as_str()))
                    .execute(&fresh)
                    .await
                    .unwrap();
            }

            let (fresh_sql,): (String,) = sqlx::query_as(
                "SELECT sql FROM sqlite_master WHERE type = 'table' AND name = 'function_calls'",
            )
            .fetch_one(&fresh)
            .await
            .unwrap();
            assert!(
                fresh_sql.to_lowercase().contains("edge_kind"),
                "fresh function_calls must carry edge_kind, got: {fresh_sql}"
            );

            // Migrated: build v29, migrate to v30.
            let dir = tempfile::tempdir().unwrap();
            let db_path = dir.path().join("test.db");
            let pool = setup_v29_schema(&db_path).await;
            let pool = migrate(pool, &db_path, 29, 30).await.unwrap();

            let (mig_sql,): (String,) = sqlx::query_as(
                "SELECT sql FROM sqlite_master WHERE type = 'table' AND name = 'function_calls'",
            )
            .fetch_one(&pool)
            .await
            .unwrap();
            assert!(
                mig_sql.to_lowercase().contains("edge_kind"),
                "migrated function_calls must carry edge_kind, got: {mig_sql}"
            );
        });
    }

    /// Build a v30-shaped DB: a `file_registry` WITHOUT the v31
    /// `parse_failed_parser_version` column, plus a `chunks` table the drift
    /// query reads. Schema version stamped 30.
    async fn setup_v30_schema(db_path: &std::path::Path) -> SqlitePool {
        let pool = SqlitePoolOptions::new()
            .max_connections(1)
            .connect_with(
                sqlx::sqlite::SqliteConnectOptions::new()
                    .filename(db_path)
                    .create_if_missing(true)
                    .foreign_keys(true),
            )
            .await
            .unwrap();
        sqlx::query("CREATE TABLE metadata (key TEXT PRIMARY KEY, value TEXT NOT NULL)")
            .execute(&pool)
            .await
            .unwrap();
        // Pre-v31 file_registry: no parse_failed_parser_version column.
        sqlx::query(
            "CREATE TABLE file_registry (
                origin TEXT PRIMARY KEY,
                source_mtime INTEGER,
                source_size INTEGER,
                source_content_hash BLOB
            )",
        )
        .execute(&pool)
        .await
        .unwrap();
        sqlx::query("INSERT INTO metadata (key, value) VALUES ('schema_version', '30')")
            .execute(&pool)
            .await
            .unwrap();
        pool
    }

    /// v30 → v31 adds the nullable `file_registry.parse_failed_parser_version`
    /// column. A pre-migration row reads back NULL for it; a fresh write can
    /// carry a version integer.
    #[test]
    fn test_migrate_v30_to_v31_adds_parse_failed_column() {
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("test.db");

        rt.block_on(async {
            let pool = setup_v30_schema(&db_path).await;

            // Pre-migration row (no marker column yet).
            sqlx::query("INSERT INTO file_registry (origin, source_mtime) VALUES ('src/a.rs', 1)")
                .execute(&pool)
                .await
                .unwrap();

            let pool = migrate(pool, &db_path, 30, 31).await.unwrap();

            // Pre-migration row reads NULL for the new column.
            let (marker,): (Option<i64>,) = sqlx::query_as(
                "SELECT parse_failed_parser_version FROM file_registry WHERE origin = 'src/a.rs'",
            )
            .fetch_one(&pool)
            .await
            .unwrap();
            assert_eq!(marker, None, "pre-v31 row must read NULL marker");

            // A fresh write can carry a version.
            sqlx::query(
                "INSERT INTO file_registry (origin, parse_failed_parser_version) \
                 VALUES ('src/b.rs', 6)",
            )
            .execute(&pool)
            .await
            .unwrap();
            let (marker2,): (Option<i64>,) = sqlx::query_as(
                "SELECT parse_failed_parser_version FROM file_registry WHERE origin = 'src/b.rs'",
            )
            .fetch_one(&pool)
            .await
            .unwrap();
            assert_eq!(marker2, Some(6));
        });
    }

    /// Fresh-create (schema.sql) and migrated (v30 → v31) must produce the same
    /// `file_registry` shape: both carry `parse_failed_parser_version`.
    #[test]
    fn test_v31_fresh_create_matches_migrated_shape() {
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();

        rt.block_on(async {
            let fresh = SqlitePoolOptions::new()
                .max_connections(1)
                .connect_with(
                    sqlx::sqlite::SqliteConnectOptions::new()
                        .filename(":memory:")
                        .foreign_keys(true),
                )
                .await
                .unwrap();
            for stmt in crate::store::split_sql_statements(include_str!("../schema.sql")) {
                if stmt.is_empty() {
                    continue;
                }
                sqlx::query(sqlx::AssertSqlSafe(stmt.as_str()))
                    .execute(&fresh)
                    .await
                    .unwrap();
            }
            let (fresh_sql,): (String,) = sqlx::query_as(
                "SELECT sql FROM sqlite_master WHERE type = 'table' AND name = 'file_registry'",
            )
            .fetch_one(&fresh)
            .await
            .unwrap();
            assert!(
                fresh_sql
                    .to_lowercase()
                    .contains("parse_failed_parser_version"),
                "fresh file_registry must carry parse_failed_parser_version, got: {fresh_sql}"
            );

            let dir = tempfile::tempdir().unwrap();
            let db_path = dir.path().join("test.db");
            let pool = setup_v30_schema(&db_path).await;
            let pool = migrate(pool, &db_path, 30, 31).await.unwrap();
            let (mig_sql,): (String,) = sqlx::query_as(
                "SELECT sql FROM sqlite_master WHERE type = 'table' AND name = 'file_registry'",
            )
            .fetch_one(&pool)
            .await
            .unwrap();
            assert!(
                mig_sql
                    .to_lowercase()
                    .contains("parse_failed_parser_version"),
                "migrated file_registry must carry parse_failed_parser_version, got: {mig_sql}"
            );
        });
    }

    /// Build a v31-shaped DB: a `function_calls` table (with the v30 edge_kind
    /// column) but NO `candidate_edges` side-table. Schema version stamped 31.
    /// Minimal — just enough for the v31→v32 step, which only CREATEs a new
    /// table and does not depend on prior shape.
    async fn setup_v31_schema(db_path: &std::path::Path) -> SqlitePool {
        let pool = SqlitePoolOptions::new()
            .max_connections(1)
            .connect_with(
                sqlx::sqlite::SqliteConnectOptions::new()
                    .filename(db_path)
                    .create_if_missing(true)
                    .foreign_keys(true),
            )
            .await
            .unwrap();
        sqlx::query("CREATE TABLE metadata (key TEXT PRIMARY KEY, value TEXT NOT NULL)")
            .execute(&pool)
            .await
            .unwrap();
        sqlx::query(
            "CREATE TABLE function_calls (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                file TEXT NOT NULL,
                caller_name TEXT NOT NULL,
                caller_line INTEGER NOT NULL,
                callee_name TEXT NOT NULL,
                call_line INTEGER NOT NULL,
                edge_kind TEXT NOT NULL DEFAULT 'call'
            )",
        )
        .execute(&pool)
        .await
        .unwrap();
        sqlx::query("INSERT INTO metadata (key, value) VALUES ('schema_version', '31')")
            .execute(&pool)
            .await
            .unwrap();
        pool
    }

    /// v31 → v32 creates the `candidate_edges` side-table. After the migration a
    /// row can be inserted and read back, and the side-table is SEPARATE from
    /// `function_calls` (a candidate is never a function_calls row).
    #[test]
    fn test_migrate_v31_to_v32_creates_candidate_edges() {
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("test.db");

        rt.block_on(async {
            let pool = setup_v31_schema(&db_path).await;

            // No candidate_edges table before the migration.
            let pre: Option<(String,)> = sqlx::query_as(
                "SELECT name FROM sqlite_master WHERE type='table' AND name='candidate_edges'",
            )
            .fetch_optional(&pool)
            .await
            .unwrap();
            assert!(pre.is_none(), "candidate_edges must not exist pre-v32");

            let pool = migrate(pool, &db_path, 31, 32).await.unwrap();

            // The table exists and a row round-trips.
            sqlx::query(
                "INSERT INTO candidate_edges (file, callee_name, ref_line, candidate_kind) \
                 VALUES ('src/a.rs', 'maybe_fn', 12, 'unresolved')",
            )
            .execute(&pool)
            .await
            .unwrap();
            let (file, callee, line, kind): (String, String, i64, String) = sqlx::query_as(
                "SELECT file, callee_name, ref_line, candidate_kind \
                 FROM candidate_edges WHERE callee_name = 'maybe_fn'",
            )
            .fetch_one(&pool)
            .await
            .unwrap();
            assert_eq!(
                (file.as_str(), callee.as_str(), line, kind.as_str()),
                ("src/a.rs", "maybe_fn", 12, "unresolved")
            );

            // It is empty after migrate (additive, no backfill).
            let pool2 = setup_v31_schema(&dir.path().join("empty.db")).await;
            let pool2 = migrate(pool2, &dir.path().join("empty.db"), 31, 32)
                .await
                .unwrap();
            let (n,): (i64,) = sqlx::query_as("SELECT COUNT(*) FROM candidate_edges")
                .fetch_one(&pool2)
                .await
                .unwrap();
            assert_eq!(n, 0, "candidate_edges must be empty on migrate");
        });
    }

    /// Fresh-create (schema.sql) and migrated (v31 → v32) must produce the same
    /// `candidate_edges` shape (same columns, same indexes). The headline
    /// `test_fresh_init_equals_full_migration_from_v10` covers this structurally
    /// too, but a focused per-step check names the exact divergent surface.
    #[test]
    fn test_v32_fresh_create_matches_migrated_shape() {
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();

        rt.block_on(async {
            // Fresh: apply schema.sql and read the candidate_edges DDL + indexes.
            let fresh = SqlitePoolOptions::new()
                .max_connections(1)
                .connect_with(
                    sqlx::sqlite::SqliteConnectOptions::new()
                        .filename(":memory:")
                        .foreign_keys(true),
                )
                .await
                .unwrap();
            for stmt in crate::store::split_sql_statements(include_str!("../schema.sql")) {
                if stmt.is_empty() {
                    continue;
                }
                sqlx::query(sqlx::AssertSqlSafe(stmt.as_str()))
                    .execute(&fresh)
                    .await
                    .unwrap();
            }
            let fresh_idx: Vec<(String,)> = sqlx::query_as(
                "SELECT name FROM sqlite_master WHERE type='index' \
                 AND tbl_name='candidate_edges' ORDER BY name",
            )
            .fetch_all(&fresh)
            .await
            .unwrap();
            let fresh_idx: Vec<String> = fresh_idx.into_iter().map(|(n,)| n).collect();
            assert!(
                fresh_idx.contains(&"idx_candidate_edges_callee".to_string())
                    && fresh_idx.contains(&"idx_candidate_edges_file".to_string()),
                "fresh candidate_edges must carry both indexes, got: {fresh_idx:?}"
            );

            // Migrated: build v31, migrate to v32, read the same.
            let dir = tempfile::tempdir().unwrap();
            let db_path = dir.path().join("test.db");
            let pool = setup_v31_schema(&db_path).await;
            let pool = migrate(pool, &db_path, 31, 32).await.unwrap();
            let mig_idx: Vec<(String,)> = sqlx::query_as(
                "SELECT name FROM sqlite_master WHERE type='index' \
                 AND tbl_name='candidate_edges' ORDER BY name",
            )
            .fetch_all(&pool)
            .await
            .unwrap();
            let mig_idx: Vec<String> = mig_idx.into_iter().map(|(n,)| n).collect();
            assert_eq!(
                fresh_idx, mig_idx,
                "fresh vs migrated candidate_edges index set must match"
            );
        });
    }

    /// FROZEN-ARTIFACT full-chain guard (legacy-state / version-null shape).
    ///
    /// Every other migration test in this module hand-builds a *minimal*
    /// predecessor schema tailored to a single step and runs at most ~3
    /// contiguous steps (the widest is `migrate(8, 11)` / `migrate(17, 19)`).
    /// NOTHING exercises the full v10→CURRENT chain over one DB, and nothing
    /// starts from the *real* on-disk v10 shape (the bespoke per-step DDL omits
    /// columns the real v10 schema carried, e.g. `chunks.source_mtime`,
    /// `chunks.window_idx`, the full v10 `notes` column set). A step that
    /// silently depends on the *real* predecessor shape — or a default backfill
    /// that's wrong for the *oldest* rows — would pass every per-step test yet
    /// crash a real user's v10 DB on upgrade. No fixture built by current code
    /// can construct a v10 DB: current `init()` writes the v32 schema. The only
    /// way to reach this shape is to freeze the historical DDL, which is what
    /// this test does — the v10 `CREATE TABLE`s are copied verbatim from git
    /// commit 73cca226 (`feat: Migrate to sqlx async SQLite + schema v10`).
    ///
    /// It also seeds an OFF-GRID note sentiment (`-0.8`) — a value only a
    /// pre-v29 binary could have written (the v29 CHECK now rejects it, and
    /// `parse_notes_str` snaps inputs before write), to confirm the v28→v29
    /// clamp rewrites legacy rows to the nearest discrete value rather than
    /// tripping its own new CHECK mid-chain.
    #[test]
    fn test_migrate_full_chain_from_real_v10_artifact() {
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("legacy_v10.db");

        rt.block_on(async {
            let pool = SqlitePoolOptions::new()
                .max_connections(1)
                .connect_with(
                    sqlx::sqlite::SqliteConnectOptions::new()
                        .filename(&db_path)
                        .create_if_missing(true),
                )
                .await
                .unwrap();

            // --- The REAL v10 schema, verbatim from commit 73cca226. ---
            // metadata
            sqlx::query(
                "CREATE TABLE IF NOT EXISTS metadata (
                    key TEXT PRIMARY KEY,
                    value TEXT NOT NULL
                )",
            )
            .execute(&pool)
            .await
            .unwrap();

            // chunks (v10 shape: source_mtime / created_at / updated_at /
            // parent_id / window_idx present; NONE of the post-v10 columns)
            sqlx::query(
                "CREATE TABLE IF NOT EXISTS chunks (
                    id TEXT PRIMARY KEY,
                    origin TEXT NOT NULL,
                    source_type TEXT NOT NULL,
                    language TEXT NOT NULL,
                    chunk_type TEXT NOT NULL,
                    name TEXT NOT NULL,
                    signature TEXT NOT NULL,
                    content TEXT NOT NULL,
                    content_hash TEXT NOT NULL,
                    doc TEXT,
                    line_start INTEGER NOT NULL,
                    line_end INTEGER NOT NULL,
                    embedding BLOB NOT NULL,
                    source_mtime INTEGER,
                    created_at TEXT NOT NULL,
                    updated_at TEXT NOT NULL,
                    parent_id TEXT,
                    window_idx INTEGER
                )",
            )
            .execute(&pool)
            .await
            .unwrap();

            sqlx::query(
                "CREATE TABLE IF NOT EXISTS calls (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    caller_id TEXT NOT NULL,
                    callee_name TEXT NOT NULL,
                    line_number INTEGER NOT NULL,
                    FOREIGN KEY (caller_id) REFERENCES chunks(id) ON DELETE CASCADE
                )",
            )
            .execute(&pool)
            .await
            .unwrap();

            sqlx::query(
                "CREATE TABLE IF NOT EXISTS function_calls (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    file TEXT NOT NULL,
                    caller_name TEXT NOT NULL,
                    caller_line INTEGER NOT NULL,
                    callee_name TEXT NOT NULL,
                    call_line INTEGER NOT NULL
                )",
            )
            .execute(&pool)
            .await
            .unwrap();

            // notes (v10 shape: NO `kind` column — that arrives at v25;
            // sentiment is a bare REAL with NO CHECK — the CHECK arrives at v29)
            sqlx::query(
                "CREATE TABLE IF NOT EXISTS notes (
                    id TEXT PRIMARY KEY,
                    text TEXT NOT NULL,
                    sentiment REAL NOT NULL,
                    mentions TEXT,
                    embedding BLOB NOT NULL,
                    source_file TEXT NOT NULL,
                    file_mtime INTEGER NOT NULL,
                    created_at TEXT NOT NULL,
                    updated_at TEXT NOT NULL
                )",
            )
            .execute(&pool)
            .await
            .unwrap();

            // --- Populate with realistic legacy rows. ---
            sqlx::query(
                "INSERT INTO chunks \
                 (id, origin, source_type, language, chunk_type, name, signature, \
                  content, content_hash, doc, line_start, line_end, embedding, \
                  source_mtime, created_at, updated_at, parent_id, window_idx) \
                 VALUES \
                 ('c1', 'src/foo.rs', 'file', 'rust', 'function', 'foo', 'fn foo()', \
                  'fn foo() {}', 'hashfoo', 'docs', 1, 3, X'00010203', \
                  1700000000, '2024-01-01T00:00:00Z', '2024-01-01T00:00:00Z', NULL, NULL)",
            )
            .execute(&pool)
            .await
            .unwrap();

            // A `calls` row pointing at the chunk (FK target exists).
            sqlx::query(
                "INSERT INTO calls (caller_id, callee_name, line_number) \
                 VALUES ('c1', 'bar', 2)",
            )
            .execute(&pool)
            .await
            .unwrap();

            sqlx::query(
                "INSERT INTO function_calls \
                 (file, caller_name, caller_line, callee_name, call_line) \
                 VALUES ('src/foo.rs', 'foo', 1, 'bar', 2)",
            )
            .execute(&pool)
            .await
            .unwrap();

            // An OFF-GRID sentiment note: only a pre-v29 binary could write
            // -0.8 (current code snaps before write AND the v29 CHECK rejects
            // it). The v28→v29 clamp must rewrite it to the nearest discrete
            // value (-1.0) so the rebuilt table satisfies its own CHECK.
            sqlx::query(
                "INSERT INTO notes \
                 (id, text, sentiment, mentions, embedding, source_file, \
                  file_mtime, created_at, updated_at) \
                 VALUES \
                 ('note:0', 'legacy off-grid note', -0.8, '[\"src/foo.rs\"]', \
                  X'', 'docs/notes.toml', 1700000000, \
                  '2024-01-01T00:00:00Z', '2024-01-01T00:00:00Z')",
            )
            .execute(&pool)
            .await
            .unwrap();

            // An on-grid note to confirm well-formed rows survive untouched.
            sqlx::query(
                "INSERT INTO notes \
                 (id, text, sentiment, mentions, embedding, source_file, \
                  file_mtime, created_at, updated_at) \
                 VALUES \
                 ('note:1', 'legacy on-grid note', 0.5, NULL, X'', \
                  'docs/notes.toml', 1700000000, \
                  '2024-01-01T00:00:00Z', '2024-01-01T00:00:00Z')",
            )
            .execute(&pool)
            .await
            .unwrap();

            sqlx::query("INSERT INTO metadata (key, value) VALUES ('schema_version', '10')")
                .execute(&pool)
                .await
                .unwrap();

            // --- Run the FULL chain v10 → CURRENT over the frozen artifact. ---
            let pool = migrate(pool, &db_path, 10, CURRENT_SCHEMA_VERSION)
                .await
                .unwrap_or_else(|e| {
                    panic!(
                        "full v10→v{CURRENT_SCHEMA_VERSION} chain over a REAL v10 artifact \
                         must succeed, got: {e:?}"
                    )
                });

            // Version stamped to current.
            let (ver,): (String,) =
                sqlx::query_as("SELECT value FROM metadata WHERE key = 'schema_version'")
                    .fetch_one(&pool)
                    .await
                    .unwrap();
            assert_eq!(
                ver,
                CURRENT_SCHEMA_VERSION.to_string(),
                "chain must land on the current version"
            );

            // Chunk data survived the chain (no row loss across the rebuilds).
            let (chunk_n,): (i64,) = sqlx::query_as("SELECT COUNT(*) FROM chunks")
                .fetch_one(&pool)
                .await
                .unwrap();
            assert_eq!(chunk_n, 1, "the single legacy chunk must survive the chain");

            let (content_hash,): (String,) =
                sqlx::query_as("SELECT content_hash FROM chunks WHERE id = 'c1'")
                    .fetch_one(&pool)
                    .await
                    .unwrap();
            assert_eq!(content_hash, "hashfoo", "chunk payload must be intact");

            // Columns added across the chain exist with correct defaults on the
            // OLDEST row (these defaults must be right for a v10 row, not just a
            // v(N-1) row).
            let (edge_kind,): (String,) =
                sqlx::query_as("SELECT edge_kind FROM function_calls LIMIT 1")
                    .fetch_one(&pool)
                    .await
                    .unwrap();
            assert_eq!(
                edge_kind, "call",
                "v30 edge_kind default must apply to v10 rows"
            );

            let (parser_version,): (i64,) =
                sqlx::query_as("SELECT parser_version FROM chunks WHERE id = 'c1'")
                    .fetch_one(&pool)
                    .await
                    .unwrap();
            assert_eq!(
                parser_version, 0,
                "v21 parser_version default must be 0 for v10 rows"
            );

            // The off-grid note was clamp-rewritten to the nearest discrete
            // value; the rebuilt table's CHECK is satisfied.
            let (sent0,): (f64,) =
                sqlx::query_as("SELECT sentiment FROM notes WHERE id = 'note:0'")
                    .fetch_one(&pool)
                    .await
                    .unwrap();
            assert_eq!(
                sent0, -1.0,
                "v29 clamp must snap the legacy -0.8 to the nearest discrete value (-1.0)"
            );

            let (sent1,): (f64,) =
                sqlx::query_as("SELECT sentiment FROM notes WHERE id = 'note:1'")
                    .fetch_one(&pool)
                    .await
                    .unwrap();
            assert_eq!(
                sent1, 0.5,
                "an on-grid legacy note must pass through unchanged"
            );

            // The v29 CHECK is actually in force on the rebuilt table.
            let insert_offgrid = sqlx::query(
                "INSERT INTO notes \
                 (id, text, sentiment, mentions, embedding, source_file, \
                  file_mtime, created_at, updated_at, kind) \
                 VALUES ('note:bad', 't', -0.3, NULL, X'', 'f', 0, 'a', 'b', NULL)",
            )
            .execute(&pool)
            .await;
            assert!(
                insert_offgrid.is_err(),
                "post-chain notes table must enforce the discrete-sentiment CHECK"
            );

            // file_registry (v29) exists and carries the v31 column.
            let (reg_sql,): (String,) = sqlx::query_as(
                "SELECT sql FROM sqlite_master WHERE type = 'table' AND name = 'file_registry'",
            )
            .fetch_one(&pool)
            .await
            .unwrap();
            assert!(
                reg_sql
                    .to_lowercase()
                    .contains("parse_failed_parser_version"),
                "file_registry must carry the v31 column after the full chain"
            );

            // candidate_edges (v32) exists and is empty after the chain (additive,
            // no backfill); a row can be inserted with its four columns.
            sqlx::query(
                "INSERT INTO candidate_edges (file, callee_name, ref_line, candidate_kind) \
                 VALUES ('src/x.rs', 'cand_fn', 3, 'unresolved')",
            )
            .execute(&pool)
            .await
            .expect("candidate_edges must exist with its four columns after the full chain");

            // Tables created mid-chain exist (type_edges v11, sparse_vectors
            // v16, llm_summaries v13/v16, candidate_edges v32). A missing one
            // means a step's `IF NOT EXISTS` masked a real ordering bug.
            for tbl in [
                "type_edges",
                "sparse_vectors",
                "llm_summaries",
                "candidate_edges",
            ] {
                let present: Option<(String,)> = sqlx::query_as(
                    "SELECT name FROM sqlite_master WHERE type='table' AND name = ?1",
                )
                .bind(tbl)
                .fetch_optional(&pool)
                .await
                .unwrap();
                assert!(present.is_some(), "{tbl} must exist after the full chain");
            }
        });
    }

    // ========================================================================
    // Part 1: fresh-init ≡ full-migration STRUCTURAL equivalence (the headline).
    //
    // The per-step `test_vN_fresh_create_matches_migrated_shape` tests above
    // each check ONE object's presence via a substring match on its `.sql`
    // text. None of them compares the WHOLE schema — the full table set, every
    // column's (name/type/notnull/default), and the full index/trigger set —
    // between a fresh `init()` DB and a DB migrated all the way from the oldest
    // supported version (v10). A migration step whose
    // `ALTER TABLE … ADD COLUMN … DEFAULT` drifts from `schema.sql`'s column
    // definition (wrong type, wrong default, missing NOT NULL) would leave
    // migrated DBs structurally different from fresh ones, and no fresh-state
    // fixture can catch it: current `init()` only ever writes the v32 schema,
    // so the migrated-from-v10 shape is unreachable except by replaying the
    // frozen historical DDL through the live migration chain.
    // ========================================================================

    /// One column's structural identity, comparable across two DBs.
    /// `(name, type, notnull, default, pk)` — the tuple `PRAGMA table_info`
    /// returns, minus `cid` (column ORDER differs between ADD COLUMN and a
    /// fresh CREATE and is not a correctness property; the *set* of columns
    /// with their definitions is).
    type ColumnDef = (String, String, i64, Option<String>, i64);

    /// One index's structural identity: `(name, is_unique, ordered column
    /// list)`. Auto-created PK/UNIQUE indexes (`sqlite_autoindex_*`) are
    /// included — they're part of the structural shape a query planner sees.
    type IndexDef = (String, bool, Vec<String>);

    /// Canonical, order-independent structural fingerprint of a SQLite schema:
    /// every table's column set, every index, every trigger, and the verbatim
    /// DDL of virtual tables (FTS5), all keyed by name so insertion/ALTER order
    /// can't perturb the comparison. Built with `BTreeMap`/sorted `Vec` so two
    /// equal schemas produce byte-identical debug output for a clean diff.
    #[derive(Debug, PartialEq, Eq)]
    struct SchemaFingerprint {
        /// table name → sorted column defs
        tables: std::collections::BTreeMap<String, Vec<ColumnDef>>,
        /// sorted index defs (name, unique, columns)
        indexes: Vec<IndexDef>,
        /// sorted trigger names
        triggers: Vec<String>,
        /// virtual-table (FTS) name → normalized DDL text
        virtual_tables: std::collections::BTreeMap<String, String>,
    }

    /// Normalize a DDL string for comparison: strip `-- line comments`,
    /// collapse all runs of whitespace to a single space, and lowercase, so
    /// cosmetic formatting differences (and the inline comments SQLite stores
    /// verbatim in `sqlite_master.sql`) between schema.sql and a migration's
    /// CREATE don't register as structural drift. Comments are prose, not a
    /// structural property; the column list / tokenizer options are.
    fn normalize_ddl(sql: &str) -> String {
        sql.lines()
            .map(|line| match line.find("--") {
                Some(i) => &line[..i],
                None => line,
            })
            .collect::<Vec<_>>()
            .join(" ")
            .split_whitespace()
            .collect::<Vec<_>>()
            .join(" ")
            .to_lowercase()
    }

    async fn fingerprint_schema(pool: &SqlitePool) -> SchemaFingerprint {
        // Regular tables (exclude SQLite internals and FTS shadow tables, which
        // are an implementation detail of the virtual table).
        let table_rows: Vec<(String,)> = sqlx::query_as(
            "SELECT name FROM sqlite_master \
             WHERE type = 'table' \
               AND name NOT LIKE 'sqlite_%' \
               AND name NOT LIKE '%_fts_%' \
               AND name NOT IN ('chunks_fts', 'notes_fts') \
             ORDER BY name",
        )
        .fetch_all(pool)
        .await
        .unwrap();

        let mut tables = std::collections::BTreeMap::new();
        for (tbl,) in &table_rows {
            // PRAGMA table_info: (cid, name, type, notnull, dflt_value, pk).
            // The table name comes from sqlite_master (not user input), so the
            // dynamic PRAGMA string is safe — AssertSqlSafe documents that.
            let cols: Vec<(i64, String, String, i64, Option<String>, i64)> =
                sqlx::query_as(sqlx::AssertSqlSafe(format!("PRAGMA table_info('{tbl}')")))
                    .fetch_all(pool)
                    .await
                    .unwrap();
            let mut defs: Vec<ColumnDef> = cols
                .into_iter()
                .map(|(_cid, name, ty, notnull, dflt, pk)| {
                    // Normalize the default's whitespace/case so e.g. `0` vs `0`
                    // and `'call'` vs `'call'` compare cleanly.
                    let dflt = dflt.map(|d| normalize_ddl(&d));
                    (name, ty.to_lowercase(), notnull, dflt, pk)
                })
                .collect();
            defs.sort();
            tables.insert(tbl.clone(), defs);
        }

        // Indexes. Enumerate via PRAGMA index_list per table to capture the
        // full structural set the planner sees, including PK/UNIQUE
        // auto-indexes that don't appear in sqlite_master.
        let mut indexes: Vec<IndexDef> = Vec::new();
        for (tbl,) in &table_rows {
            // PRAGMA index_list: (seq, name, unique, origin, partial). Table /
            // index names originate from the catalog, not user input.
            let idx_list: Vec<(i64, String, i64, String, i64)> =
                sqlx::query_as(sqlx::AssertSqlSafe(format!("PRAGMA index_list('{tbl}')")))
                    .fetch_all(pool)
                    .await
                    .unwrap();
            for (_seq, name, unique, _origin, _partial) in idx_list {
                // PRAGMA index_info: (seqno, cid, name) — name is the column;
                // NULL for expression columns (none in our schema).
                let cols: Vec<(i64, i64, Option<String>)> =
                    sqlx::query_as(sqlx::AssertSqlSafe(format!("PRAGMA index_info('{name}')")))
                        .fetch_all(pool)
                        .await
                        .unwrap();
                let col_names: Vec<String> = cols
                    .into_iter()
                    .map(|(_s, _c, n)| n.unwrap_or_else(|| "<expr>".to_string()))
                    .collect();
                indexes.push((name, unique != 0, col_names));
            }
        }
        indexes.sort();

        // Triggers.
        let trigger_rows: Vec<(String,)> =
            sqlx::query_as("SELECT name FROM sqlite_master WHERE type = 'trigger' ORDER BY name")
                .fetch_all(pool)
                .await
                .unwrap();
        let triggers: Vec<String> = trigger_rows.into_iter().map(|(n,)| n).collect();

        // Virtual tables (FTS5): PRAGMA table_info is unreliable for these, so
        // compare the normalized CREATE VIRTUAL TABLE DDL directly.
        let vtab_rows: Vec<(String, String)> = sqlx::query_as(
            "SELECT name, sql FROM sqlite_master \
             WHERE type = 'table' AND sql LIKE '%VIRTUAL TABLE%' \
             ORDER BY name",
        )
        .fetch_all(pool)
        .await
        .unwrap();
        let mut virtual_tables = std::collections::BTreeMap::new();
        for (name, sql) in vtab_rows {
            virtual_tables.insert(name, normalize_ddl(&sql));
        }

        SchemaFingerprint {
            tables,
            indexes,
            triggers,
            virtual_tables,
        }
    }

    /// Apply `schema.sql` to a fresh file-backed DB exactly the way
    /// `Store::init` does (BEGIN/END-aware split), returning the open pool.
    async fn fresh_init_schema(db_path: &std::path::Path) -> SqlitePool {
        let pool = SqlitePoolOptions::new()
            .max_connections(1)
            .connect_with(
                sqlx::sqlite::SqliteConnectOptions::new()
                    .filename(db_path)
                    .create_if_missing(true)
                    .foreign_keys(true),
            )
            .await
            .unwrap();
        for stmt in crate::store::split_sql_statements(include_str!("../schema.sql")) {
            if stmt.is_empty() {
                continue;
            }
            sqlx::query(sqlx::AssertSqlSafe(stmt.as_str()))
                .execute(&pool)
                .await
                .unwrap();
        }
        // init() also stamps schema_version; mirror it so the DB is in a
        // realistic post-init state (the structural comparison ignores rows).
        sqlx::query("INSERT OR REPLACE INTO metadata (key, value) VALUES ('schema_version', ?1)")
            .bind(CURRENT_SCHEMA_VERSION.to_string())
            .execute(&pool)
            .await
            .unwrap();
        pool
    }

    /// Build the REAL v10 on-disk schema (verbatim DDL from the original v10
    /// schema, the same frozen artifact `test_migrate_full_chain_from_real_v10_artifact`
    /// uses) at `db_path`, stamped at version 10. Returns the open pool ready
    /// for `migrate(10, CURRENT)`. A real v10 DB carries all the v10 tables,
    /// indexes, and FTS virtual tables; a faithful artifact must too, or the
    /// structural comparison against fresh-init would falsely flag the migration.
    async fn build_real_v10_schema(db_path: &std::path::Path) -> SqlitePool {
        let pool = SqlitePoolOptions::new()
            .max_connections(1)
            .connect_with(
                sqlx::sqlite::SqliteConnectOptions::new()
                    .filename(db_path)
                    .create_if_missing(true)
                    .foreign_keys(true),
            )
            .await
            .unwrap();

        sqlx::query(
            "CREATE TABLE IF NOT EXISTS metadata (key TEXT PRIMARY KEY, value TEXT NOT NULL)",
        )
        .execute(&pool)
        .await
        .unwrap();
        sqlx::query(
            "CREATE TABLE IF NOT EXISTS chunks (
                id TEXT PRIMARY KEY,
                origin TEXT NOT NULL,
                source_type TEXT NOT NULL,
                language TEXT NOT NULL,
                chunk_type TEXT NOT NULL,
                name TEXT NOT NULL,
                signature TEXT NOT NULL,
                content TEXT NOT NULL,
                content_hash TEXT NOT NULL,
                doc TEXT,
                line_start INTEGER NOT NULL,
                line_end INTEGER NOT NULL,
                embedding BLOB NOT NULL,
                source_mtime INTEGER,
                created_at TEXT NOT NULL,
                updated_at TEXT NOT NULL,
                parent_id TEXT,
                window_idx INTEGER
            )",
        )
        .execute(&pool)
        .await
        .unwrap();
        sqlx::query(
            "CREATE TABLE IF NOT EXISTS calls (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                caller_id TEXT NOT NULL,
                callee_name TEXT NOT NULL,
                line_number INTEGER NOT NULL,
                FOREIGN KEY (caller_id) REFERENCES chunks(id) ON DELETE CASCADE
            )",
        )
        .execute(&pool)
        .await
        .unwrap();
        sqlx::query(
            "CREATE TABLE IF NOT EXISTS function_calls (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                file TEXT NOT NULL,
                caller_name TEXT NOT NULL,
                caller_line INTEGER NOT NULL,
                callee_name TEXT NOT NULL,
                call_line INTEGER NOT NULL
            )",
        )
        .execute(&pool)
        .await
        .unwrap();
        sqlx::query(
            "CREATE TABLE IF NOT EXISTS notes (
                id TEXT PRIMARY KEY,
                text TEXT NOT NULL,
                sentiment REAL NOT NULL,
                mentions TEXT,
                embedding BLOB NOT NULL,
                source_file TEXT NOT NULL,
                file_mtime INTEGER NOT NULL,
                created_at TEXT NOT NULL,
                updated_at TEXT NOT NULL
            )",
        )
        .execute(&pool)
        .await
        .unwrap();
        // v10 carried the two FTS virtual tables; no migration step ever ALTERs
        // them and schema.sql creates them with IF NOT EXISTS, so build BOTH
        // here verbatim from schema.sql to keep the virtual-table comparison
        // meaningful (their shapes are frozen since v10).
        sqlx::query(
            "CREATE VIRTUAL TABLE IF NOT EXISTS chunks_fts USING fts5(
                id UNINDEXED, name, signature, content, doc, tokenize='unicode61'
            )",
        )
        .execute(&pool)
        .await
        .unwrap();
        sqlx::query(
            "CREATE VIRTUAL TABLE IF NOT EXISTS notes_fts USING fts5(
                id UNINDEXED, text, tokenize='unicode61'
            )",
        )
        .execute(&pool)
        .await
        .unwrap();

        // The v10 indexes (verbatim from the original v10 schema). A real v10
        // DB carried ALL of these; no migration step recreates them, so a
        // faithful v10 artifact MUST include them or the structural comparison
        // against fresh-init would falsely flag the migration. These survive
        // every rebuild step untouched (the chunks/calls/function_calls tables
        // are never DROP/RENAME'd across the chain).
        for idx_ddl in [
            "CREATE INDEX IF NOT EXISTS idx_chunks_origin ON chunks(origin)",
            "CREATE INDEX IF NOT EXISTS idx_chunks_source_type ON chunks(source_type)",
            "CREATE INDEX IF NOT EXISTS idx_chunks_content_hash ON chunks(content_hash)",
            "CREATE INDEX IF NOT EXISTS idx_chunks_name ON chunks(name)",
            "CREATE INDEX IF NOT EXISTS idx_chunks_language ON chunks(language)",
            "CREATE INDEX IF NOT EXISTS idx_chunks_parent ON chunks(parent_id)",
            "CREATE INDEX IF NOT EXISTS idx_calls_caller ON calls(caller_id)",
            "CREATE INDEX IF NOT EXISTS idx_calls_callee ON calls(callee_name)",
            "CREATE INDEX IF NOT EXISTS idx_fcalls_file ON function_calls(file)",
            "CREATE INDEX IF NOT EXISTS idx_fcalls_caller ON function_calls(caller_name)",
            "CREATE INDEX IF NOT EXISTS idx_fcalls_callee ON function_calls(callee_name)",
            "CREATE INDEX IF NOT EXISTS idx_notes_sentiment ON notes(sentiment)",
        ] {
            sqlx::query(idx_ddl).execute(&pool).await.unwrap();
        }

        sqlx::query("INSERT INTO metadata (key, value) VALUES ('schema_version', '10')")
            .execute(&pool)
            .await
            .unwrap();
        pool
    }

    /// HEADLINE GUARD: a fresh `init()` DB and a DB migrated the full chain
    /// from the oldest supported version (v10) must be STRUCTURALLY IDENTICAL —
    /// same tables, same columns (name/type/notnull/default), same indexes,
    /// same triggers, same virtual-table DDL.
    ///
    /// This is also a SWEEP over every `ADD COLUMN … DEFAULT` migration step:
    /// each must mirror its `schema.sql` counterpart exactly. A column whose
    /// default or type differs between the migrate path and the fresh path is
    /// the straggler this catches — it would silently diverge migrated DBs from
    /// fresh ones, and no born-at-HEAD fixture can construct the migrated shape
    /// (current `init()` only ever writes v32).
    ///
    /// Calibration (proving the guard bites): temporarily changing the v30
    /// migration's `edge_kind` default from `'call'` to `'xcall'`, OR
    /// `schema.sql`'s `needs_embedding` default from `0` to `1`, makes the two
    /// fingerprints diverge and this test fails with a precise column diff.
    #[test]
    fn test_fresh_init_equals_full_migration_from_v10() {
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        rt.block_on(async {
            let fresh_dir = tempfile::tempdir().unwrap();
            let fresh_path = fresh_dir.path().join("fresh.db");
            let fresh_pool = fresh_init_schema(&fresh_path).await;
            let fresh_fp = fingerprint_schema(&fresh_pool).await;

            let mig_dir = tempfile::tempdir().unwrap();
            let mig_path = mig_dir.path().join("migrated_from_v10.db");
            let v10_pool = build_real_v10_schema(&mig_path).await;
            let mig_pool = migrate(v10_pool, &mig_path, 10, CURRENT_SCHEMA_VERSION)
                .await
                .unwrap_or_else(|e| {
                    panic!("full v10→v{CURRENT_SCHEMA_VERSION} chain must succeed, got: {e:?}")
                });
            let mig_fp = fingerprint_schema(&mig_pool).await;

            // Field-by-field comparison with targeted messages so a failure
            // names the exact divergent surface, not just "not equal".
            assert_eq!(
                fresh_fp.tables.keys().collect::<Vec<_>>(),
                mig_fp.tables.keys().collect::<Vec<_>>(),
                "table SET differs between fresh init and full migration"
            );
            for (tbl, fresh_cols) in &fresh_fp.tables {
                let mig_cols = mig_fp.tables.get(tbl).unwrap();
                assert_eq!(
                    fresh_cols, mig_cols,
                    "table `{tbl}` column defs differ (name/type/notnull/default/pk).\n\
                     fresh init: {fresh_cols:#?}\n\
                     migrated  : {mig_cols:#?}"
                );
            }
            assert_eq!(
                fresh_fp.indexes, mig_fp.indexes,
                "index set differs between fresh init and full migration.\n\
                 fresh: {:#?}\nmigrated: {:#?}",
                fresh_fp.indexes, mig_fp.indexes
            );
            assert_eq!(
                fresh_fp.triggers, mig_fp.triggers,
                "trigger set differs between fresh init and full migration"
            );
            assert_eq!(
                fresh_fp.virtual_tables, mig_fp.virtual_tables,
                "virtual-table (FTS) DDL differs between fresh init and full migration"
            );

            // Whole-fingerprint equality as the backstop — anything the
            // field-by-field checks above missed still fails here.
            assert_eq!(
                fresh_fp, mig_fp,
                "fresh-init schema and full-v10-migration schema must be structurally identical"
            );
        });
    }

    // ========================================================================
    // Part 2: the notes_fts / notes rebuild seam (v28→v29).
    //
    // The v28→v29 migration DROPs and recreates the `notes` table (CREATE
    // notes_v29 → INSERT…SELECT → DROP notes → RENAME), which assigns the
    // rebuilt `notes` rows FRESH internal rowids, while the manually-managed
    // `notes_fts` virtual table is left untouched and keeps its OLD rowids.
    //
    // Finding: this is SOUND, but NOT because "rowids line up" — they do not.
    // It is sound because NOTHING in the codebase ever couples `notes` to
    // `notes_fts` by rowid. `notes_fts` is declared `id UNINDEXED` and every
    // single access keys on `id` (the three writers in store/notes.rs:
    // INSERT (id,text), DELETE WHERE id IN (...), DELETE WHERE id IN (SELECT
    // id FROM notes ...)); and there is NO reader at all — `notes_fts` is
    // never queried via MATCH anywhere in the tree or its git history. The
    // v28→v29 rebuild preserves every `id` verbatim, so the id-coupling the
    // writers depend on is intact across the rebuild even though the rowids
    // are not.
    //
    // The guard below pins that id-coupling invariant across the rebuild: it
    // builds a v28 notes table with seeded rows + matching FTS rows, runs the
    // real migration, and asserts (a) every `notes.id` still id-joins to
    // exactly one matching `notes_fts` row carrying the right text, (b) the
    // rowids deliberately do NOT line up (documenting WHY id-coupling is the
    // load-bearing invariant), and (c) the id-keyed delete the writers issue
    // still removes the correct FTS row. If a future change introduces a
    // rowid-based notes_fts read, or a migration drops/recreates `notes_fts`
    // and loses the id rows, this goes red.
    // ========================================================================

    /// Build a v28-shaped DB: a `notes` table WITHOUT the v29 sentiment CHECK
    /// (bare REAL), plus the `notes_fts` virtual table, stamped at version 28.
    async fn setup_v28_notes_schema(db_path: &std::path::Path) -> SqlitePool {
        let pool = SqlitePoolOptions::new()
            .max_connections(1)
            .connect_with(
                sqlx::sqlite::SqliteConnectOptions::new()
                    .filename(db_path)
                    .create_if_missing(true)
                    .foreign_keys(true),
            )
            .await
            .unwrap();
        sqlx::query("CREATE TABLE metadata (key TEXT PRIMARY KEY, value TEXT NOT NULL)")
            .execute(&pool)
            .await
            .unwrap();
        // v28 notes: bare REAL sentiment (no CHECK), `kind` column present
        // (arrived v25). Matches the pre-v29 on-disk shape.
        sqlx::query(
            "CREATE TABLE notes (
                id TEXT PRIMARY KEY,
                text TEXT NOT NULL,
                sentiment REAL NOT NULL,
                mentions TEXT,
                embedding BLOB NOT NULL,
                source_file TEXT NOT NULL,
                file_mtime INTEGER NOT NULL,
                created_at TEXT NOT NULL,
                updated_at TEXT NOT NULL,
                kind TEXT
            )",
        )
        .execute(&pool)
        .await
        .unwrap();
        sqlx::query("CREATE INDEX idx_notes_sentiment ON notes(sentiment)")
            .execute(&pool)
            .await
            .unwrap();
        sqlx::query("CREATE INDEX idx_notes_kind ON notes(kind)")
            .execute(&pool)
            .await
            .unwrap();
        sqlx::query(
            "CREATE VIRTUAL TABLE notes_fts USING fts5(id UNINDEXED, text, tokenize='unicode61')",
        )
        .execute(&pool)
        .await
        .unwrap();
        sqlx::query("INSERT INTO metadata (key, value) VALUES ('schema_version', '28')")
            .execute(&pool)
            .await
            .unwrap();
        pool
    }

    /// SEAM GUARD: the v28→v29 `notes`-table rebuild must keep
    /// `notes_fts` id-coupled to the rebuilt `notes` rows. Asserts that across
    /// the rebuild every `notes.id` retains exactly one matching `notes_fts`
    /// row with the correct text (the id-join every writer depends on), that the
    /// id-keyed FTS delete still targets the right row, and documents that the
    /// rowids deliberately do NOT line up (so the invariant is the `id`
    /// coupling, not rowid identity).
    #[test]
    fn test_v28_to_v29_notes_fts_id_coupling_survives_rebuild() {
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("v28_notes.db");
        rt.block_on(async {
            let pool = setup_v28_notes_schema(&db_path).await;

            // Seed notes + matching FTS rows.
            for (id, text, sent) in [
                ("note:0", "alpha load bearing", -0.5_f64),
                ("note:1", "beta important", 0.5),
                ("note:2", "gamma critical", 1.0),
            ] {
                sqlx::query(
                    "INSERT INTO notes (id, text, sentiment, mentions, embedding, \
                     source_file, file_mtime, created_at, updated_at, kind) \
                     VALUES (?1, ?2, ?3, NULL, X'', 'docs/notes.toml', 0, 'a', 'b', NULL)",
                )
                .bind(id)
                .bind(text)
                .bind(sent)
                .execute(&pool)
                .await
                .unwrap();
                sqlx::query("INSERT INTO notes_fts (id, text) VALUES (?1, ?2)")
                    .bind(id)
                    .bind(text)
                    .execute(&pool)
                    .await
                    .unwrap();
            }
            // Churn note:1's FTS row so its FTS internal rowid is reassigned
            // past the others — guarantees notes.rowid != notes_fts.rowid for
            // it, so the read path provably cannot rely on rowid equality.
            sqlx::query("DELETE FROM notes_fts WHERE id = 'note:1'")
                .execute(&pool)
                .await
                .unwrap();
            sqlx::query("INSERT INTO notes_fts (id, text) VALUES ('note:1', 'beta important')")
                .execute(&pool)
                .await
                .unwrap();

            // Run the real v28→v29 migration (rebuilds the notes table).
            let pool = migrate(pool, &db_path, 28, 29).await.unwrap();

            // (a) Every notes.id id-joins to exactly one notes_fts row with the
            //     correct text — the id-join is intact across the rebuild.
            for (id, want_text) in [
                ("note:0", "alpha load bearing"),
                ("note:1", "beta important"),
                ("note:2", "gamma critical"),
            ] {
                let rows: Vec<(String,)> = sqlx::query_as(
                    "SELECT f.text FROM notes n JOIN notes_fts f ON n.id = f.id WHERE n.id = ?1",
                )
                .bind(id)
                .fetch_all(&pool)
                .await
                .unwrap();
                assert_eq!(
                    rows.len(),
                    1,
                    "id `{id}` must id-join to exactly one notes_fts row after rebuild"
                );
                assert_eq!(
                    rows[0].0, want_text,
                    "notes_fts text for id `{id}` must match the source note text \
                     (no stale/mismatched FTS content across the rebuild)"
                );
            }

            // (b) The rowids deliberately do NOT line up after the rebuild —
            //     the rebuilt notes table got fresh rowids while notes_fts kept
            //     its old ones. This documents WHY id-coupling (not rowid
            //     identity) is the load-bearing invariant.
            let mismatched: Vec<(String, i64, i64)> = sqlx::query_as(
                "SELECT n.id, n.rowid, f.rowid \
                 FROM notes n JOIN notes_fts f ON n.id = f.id \
                 WHERE n.rowid != f.rowid",
            )
            .fetch_all(&pool)
            .await
            .unwrap();
            assert!(
                !mismatched.is_empty(),
                "expected at least one id whose notes.rowid != notes_fts.rowid after the \
                 rebuild — if they line up, this test no longer proves the id-coupling \
                 (vs rowid) is what makes the seam sound"
            );

            // (c) The id-keyed FTS delete the writers issue still removes the
            //     correct row (and only it) post-rebuild.
            sqlx::query("DELETE FROM notes_fts WHERE id IN ('note:1')")
                .execute(&pool)
                .await
                .unwrap();
            let (remaining,): (i64,) =
                sqlx::query_as("SELECT COUNT(*) FROM notes_fts WHERE id = 'note:1'")
                    .fetch_one(&pool)
                    .await
                    .unwrap();
            assert_eq!(remaining, 0, "id-keyed FTS delete must remove note:1's row");
            let (others,): (i64,) =
                sqlx::query_as("SELECT COUNT(*) FROM notes_fts WHERE id IN ('note:0', 'note:2')")
                    .fetch_one(&pool)
                    .await
                    .unwrap();
            assert_eq!(others, 2, "id-keyed delete must not touch the other rows");
        });
    }
}
