clawhdf5/crates/clawhdf5-migrate/src/main.rs

mod sqlite_reader;
mod store_writer;
mod validate;

use std::path::Path;

use clap::Parser;

use sqlite_reader::SchemaConfig;
use store_writer::Migration;
use validate::ValidationSummary;

type BoxErr = Box<dyn std::error::Error>;

/// Migrate ZeroClaw agent memory from SQLite to a clawhdf5-agent store.
///
/// The output is an ordinary agent store: open it with
/// `HDF5Memory::open` (or `clawhdf5-cli --path <store> ...`).
#[derive(Parser, Debug)]
#[command(name = "clawhdf5-migrate", version, about)]
struct Cli {
    /// Source SQLite database path
    #[arg(long)]
    sqlite: String,

    /// Destination agent store (.h5). Replaced if it exists, unless
    /// --incremental
    #[arg(long)]
    hdf5: String,

    /// Agent ID recorded in the new store
    #[arg(long, default_value = "migrated")]
    agent_id: String,

    /// Embedder name recorded in the new store
    #[arg(long, default_value = "unknown")]
    embedder: String,

    /// Embedding dimension (default: detected from the first row). Every row
    /// must have it: a row of another length is an error, never truncated.
    /// Required when the source has no memory records, so the new store can
    /// take records later
    #[arg(long)]
    embedding_dim: Option<usize>,

    /// Skip deleted/tombstoned entries (otherwise they are migrated as
    /// deleted records)
    #[arg(long)]
    skip_deleted: bool,

    /// Compress the embeddings dataset (deflate)
    #[arg(long)]
    compression: bool,

    /// Compression level 1-9
    #[arg(long, default_value_t = 4)]
    compression_level: u32,

    /// Store embeddings as full-precision f32 instead of the default half
    /// precision (float16: half the bytes, about three significant digits,
    /// values within ±65504)
    #[arg(long)]
    f32: bool,

    /// Accepted for compatibility; float16 is now the default
    #[arg(long, hide = true, conflicts_with = "f32")]
    float16: bool,

    /// Validate without writing
    #[arg(long)]
    dry_run: bool,

    /// Content-check every migrated row (default: a representative sample)
    #[arg(long)]
    validate_full: bool,

    /// Add to the store at --hdf5 if it exists, writing only source rows it
    /// does not already hold (records matched by content, sessions by id,
    /// entities by name and type). Matched records take the source's deleted
    /// flag: deleted in the source tombstones the record, active in the source
    /// writes a deleted record again. The source must have the store's
    /// embedding dimension
    #[arg(long)]
    incremental: bool,

    /// Override the SQLite table name for memory chunks
    #[arg(long)]
    chunks_table: Option<String>,

    /// Override the SQLite table name for sessions
    #[arg(long)]
    sessions_table: Option<String>,

    /// Override the SQLite table name for entities
    #[arg(long)]
    entities_table: Option<String>,

    /// Override the SQLite table name for relations
    #[arg(long)]
    relations_table: Option<String>,

    /// Print progress
    #[arg(long)]
    verbose: bool,
}

/// Build the schema config from CLI table-name overrides (defaults otherwise).
fn schema_from_cli(cli: &Cli) -> SchemaConfig {
    let mut c = SchemaConfig::default();
    if let Some(t) = &cli.chunks_table {
        c.chunks.table = t.clone();
    }
    if let Some(t) = &cli.sessions_table {
        c.sessions.table = t.clone();
    }
    if let Some(t) = &cli.entities_table {
        c.entities.table = t.clone();
    }
    if let Some(t) = &cli.relations_table {
        c.relations.table = t.clone();
    }
    c
}

/// The result of a (non-dry) run.
#[derive(Debug)]
struct Outcome {
    migration: Migration,
    summary: ValidationSummary,
}

fn run(cli: &Cli) -> Result<Option<Outcome>, BoxErr> {
    let schema = schema_from_cli(cli);

    // Dry run: a fast count-only pass that does not buffer the database.
    if cli.dry_run {
        let counts = sqlite_reader::read_counts(&cli.sqlite, cli.skip_deleted, &schema)?;
        eprintln!("Dry run — no output file written.");
        eprintln!(
            "Would migrate: {} chunks, {} sessions, {} entities, {} relations",
            counts.chunks, counts.sessions, counts.entities, counts.relations
        );
        return Ok(None);
    }

    let out = Path::new(&cli.hdf5);
    if cli.embedding_dim == Some(0) {
        return Err("--embedding-dim must be at least 1".into());
    }
    // The source's dimension is its own (--embedding-dim or the first row),
    // never the existing store's: a mismatch must be an error, not a reason
    // to reshape the source.
    //
    // An incremental run reads deleted rows even with --skip-deleted, so a
    // row deleted in the source since the last run tombstones its store
    // record; the writer still leaves out deleted rows the store lacks.
    let appending = cli.incremental && out.exists();
    let reader_skip_deleted = cli.skip_deleted && !appending;
    if cli.verbose {
        eprintln!("Reading SQLite database: {}", cli.sqlite);
    }
    let data =
        sqlite_reader::read_sqlite(&cli.sqlite, reader_skip_deleted, cli.embedding_dim, &schema)?;
    if cli.verbose {
        eprintln!(
            "Read {} chunks, {} sessions, {} entities, {} relations (dim={})",
            data.chunks.len(),
            data.sessions.len(),
            data.entities.len(),
            data.relations.len(),
            data.embedding_dim
        );
        eprintln!("Writing agent store: {}", cli.hdf5);
    }

    let opts = store_writer::WriteOptions {
        agent_id: cli.agent_id.clone(),
        embedder: cli.embedder.clone(),
        compression: cli.compression,
        compression_level: cli.compression_level.clamp(1, 9),
        f32: cli.f32,
        incremental: cli.incremental,
        skip_deleted: cli.skip_deleted,
    };
    let migration = store_writer::write_store(out, &data, &opts)?;

    if migration.appended_to_existing && cli.f32 && migration.float16 {
        eprintln!(
            "warning: --f32 ignored: the existing store is float16, and a store keeps the \
             precision it was created with"
        );
    }
    if !migration.dangling_relations.is_empty() {
        eprintln!(
            "warning: skipped {} relation(s) naming an entity id not in the entities table",
            migration.dangling_relations.len()
        );
    }
    if !migration.deleted_in_store.is_empty() || migration.restored > 0 {
        eprintln!(
            "Incremental: {} store record(s) deleted and {} restored (written again) to match \
             the source's deleted flags",
            migration.deleted_in_store.len(),
            migration.restored
        );
    }
    if cli.verbose {
        if migration.appended_to_existing {
            eprintln!(
                "Incremental: already in the store: {} records, {} sessions, {} entities, {} relations",
                migration.chunks_present,
                migration.sessions_present,
                migration.entities_present,
                migration.relations_present
            );
        }
        if !migration.anomaly_alerts.is_empty() {
            eprintln!(
                "Note: the agent's write-anomaly detector raised {} alert(s) during the import \
                 (informational; nothing was blocked), e.g.:",
                migration.anomaly_alerts.len()
            );
            for msg in migration.anomaly_alerts.iter().take(3) {
                eprintln!("  {msg}");
            }
        }
        eprintln!("Validating output (reading it back with HDF5Memory::open_read_only)...");
    }

    let summary = validate::validate_store(out, &data, &migration, cli.validate_full)?;
    Ok(Some(Outcome { migration, summary }))
}

fn main() -> Result<(), BoxErr> {
    let cli = Cli::parse();
    let Some(Outcome { migration, summary }) = run(&cli)? else {
        return Ok(());
    };
    eprintln!(
        "Migration complete: wrote {} records, {} sessions, {} entities, {} relations. \
         Store: {} records ({} active), {} sessions, {} entities, {} relations, dim={}, {}; \
         {} rows content-verified{}",
        migration.records.len(),
        migration.sessions.len(),
        migration.entities.len(),
        migration.relations.len(),
        summary.count,
        summary.active,
        summary.sessions,
        summary.entities,
        summary.relations,
        summary.embedding_dim,
        if summary.float16 { "float16" } else { "f32" },
        summary.rows_checked,
        if summary.search_checked {
            ", search verified"
        } else {
            ""
        },
    );
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;
    use clawhdf5_agent::{AgentMemory, HDF5Memory, SearchOptions};
    use clawhdf5_format::float16::round_to_f16;
    use rusqlite::Connection;
    use std::path::PathBuf;
    use tempfile::TempDir;

    /// Create a test SQLite database with the ZeroClaw schema.
    fn create_test_db(dir: &TempDir) -> String {
        let db_path = dir.path().join("test.db");
        let path_str = db_path.to_str().unwrap().to_string();
        let conn = Connection::open(&path_str).unwrap();
        conn.execute_batch(
            "CREATE TABLE memory_chunks (
                id INTEGER PRIMARY KEY,
                chunk TEXT NOT NULL,
                embedding BLOB NOT NULL,
                source_channel TEXT DEFAULT 'api',
                timestamp REAL NOT NULL,
                session_id TEXT,
                tags TEXT DEFAULT '',
                deleted INTEGER DEFAULT 0
            );
            CREATE TABLE sessions (
                id TEXT PRIMARY KEY,
                start_idx INTEGER,
                end_idx INTEGER,
                channel TEXT,
                timestamp REAL,
                summary TEXT
            );
            CREATE TABLE entities (
                id INTEGER PRIMARY KEY,
                name TEXT NOT NULL,
                type TEXT NOT NULL,
                embedding_idx INTEGER DEFAULT -1
            );
            CREATE TABLE relations (
                src INTEGER NOT NULL,
                tgt INTEGER NOT NULL,
                relation TEXT NOT NULL,
                weight REAL DEFAULT 1.0,
                timestamp REAL,
                FOREIGN KEY (src) REFERENCES entities(id),
                FOREIGN KEY (tgt) REFERENCES entities(id)
            );",
        )
        .unwrap();
        path_str
    }

    /// Insert a memory chunk with a known embedding.
    fn insert_chunk(conn: &Connection, id: i64, text: &str, embedding: &[f32], deleted: i32) {
        let blob: Vec<u8> = embedding.iter().flat_map(|v| v.to_le_bytes()).collect();
        conn.execute(
            "INSERT INTO memory_chunks (id, chunk, embedding, source_channel, timestamp, session_id, tags, deleted)
             VALUES (?1, ?2, ?3, 'api', 1700000000.0, 'sess-1', 'tag1,tag2', ?4)",
            rusqlite::params![id, text, blob, deleted],
        )
        .unwrap();
    }

    fn insert_session(conn: &Connection, id: &str, start: i64, end: i64) {
        conn.execute(
            "INSERT INTO sessions (id, start_idx, end_idx, channel, timestamp, summary)
             VALUES (?1, ?2, ?3, 'discord', 1700000000.0, 'test summary')",
            rusqlite::params![id, start, end],
        )
        .unwrap();
    }

    fn insert_entity(conn: &Connection, id: i64, name: &str, etype: &str) {
        conn.execute(
            "INSERT INTO entities (id, name, type, embedding_idx) VALUES (?1, ?2, ?3, -1)",
            rusqlite::params![id, name, etype],
        )
        .unwrap();
    }

    fn insert_relation(conn: &Connection, src: i64, tgt: i64, rel: &str) {
        conn.execute(
            "INSERT INTO relations (src, tgt, relation, weight, timestamp)
             VALUES (?1, ?2, ?3, 1.0, 1700000000.0)",
            rusqlite::params![src, tgt, rel],
        )
        .unwrap();
    }

    fn make_embedding(dim: usize, seed: f32) -> Vec<f32> {
        (0..dim).map(|i| seed + i as f32 * 0.1).collect()
    }

    /// A unit-length embedding pointing mostly along axis `axis`, so records
    /// are distinguishable by vector search.
    fn axis_embedding(dim: usize, axis: usize) -> Vec<f32> {
        let mut v: Vec<f32> = (0..dim).map(|i| 0.01 * (i as f32 + 1.0)).collect();
        v[axis % dim] = 1.0;
        let n = v.iter().map(|x| x * x).sum::<f32>().sqrt();
        v.iter().map(|x| x / n).collect()
    }

    fn out_path(dir: &TempDir, name: &str) -> PathBuf {
        dir.path().join(name)
    }

    /// Run the CLI exactly as `clawhdf5-migrate --sqlite <db> --hdf5 <out> <extra...>`.
    fn migrate(db: &str, out: &Path, extra: &[&str]) -> Result<Option<Outcome>, BoxErr> {
        let mut args = vec![
            "clawhdf5-migrate",
            "--sqlite",
            db,
            "--hdf5",
            out.to_str().unwrap(),
        ];
        args.extend_from_slice(extra);
        run(&Cli::try_parse_from(args)?)
    }

    fn migrate_ok(db: &str, out: &Path, extra: &[&str]) -> Outcome {
        migrate(db, out, extra).unwrap().expect("not a dry run")
    }

    fn write(db: &str, out: &Path, opts_f32: bool) -> (sqlite_reader::SqliteData, Migration) {
        let data = sqlite_reader::read_sqlite(db, false, None, &SchemaConfig::default()).unwrap();
        let opts = store_writer::WriteOptions {
            agent_id: "t".into(),
            embedder: "t".into(),
            compression: false,
            compression_level: 4,
            f32: opts_f32,
            incremental: false,
            skip_deleted: false,
        };
        let m = store_writer::write_store(out, &data, &opts).unwrap();
        (data, m)
    }

    // ---------- Basic end-to-end migration ----------
    #[test]
    fn test_basic_migration() {
        let dir = TempDir::new().unwrap();
        let db_path = create_test_db(&dir);
        let h5_path = out_path(&dir, "out.h5");

        let conn = Connection::open(&db_path).unwrap();
        insert_chunk(&conn, 1, "hello world", &make_embedding(8, 1.0), 0);
        insert_chunk(&conn, 2, "goodbye world", &make_embedding(8, 2.0), 0);
        insert_session(&conn, "s1", 0, 1);
        insert_entity(&conn, 1, "Alice", "person");
        insert_relation(&conn, 1, 1, "self");
        drop(conn);

        let o = migrate_ok(&db_path, &h5_path, &["--agent-id", "test-agent"]);
        assert_eq!(o.summary.count, 2);
        assert_eq!(o.summary.sessions, 1);
        assert_eq!(o.summary.entities, 1);
        assert_eq!(o.summary.relations, 1);
        assert_eq!(o.summary.embedding_dim, 8);
        assert!(o.summary.float16, "float16 is the default");
        assert!(o.summary.search_checked);
    }

    // ---------- The migrated file is a real agent store ----------

    /// Build a source with distinct records, sessions and a small graph,
    /// migrate it with `extra` flags, and use the result as an agent would.
    fn end_to_end(extra: &[&str], expect_float16: bool) {
        let dir = TempDir::new().unwrap();
        let db_path = create_test_db(&dir);
        let h5_path = out_path(&dir, "agent.h5");
        let dim = 16;
        let texts = [
            "the deploy key rotates every ninety days",
            "alice prefers tea over coffee in the morning",
            "the build cache lives on the tank runner",
            "bob is allergic to peanuts",
            "quarterly review is scheduled for october",
        ];

        let conn = Connection::open(&db_path).unwrap();
        for (i, t) in texts.iter().enumerate() {
            insert_chunk(&conn, i as i64 + 1, t, &axis_embedding(dim, i * 3), 0);
        }
        insert_chunk(&conn, 99, "a forgotten memory", &axis_embedding(dim, 15), 1);
        conn.execute(
            "INSERT INTO sessions VALUES ('sess-a', 0, 2, 'discord', 1700000123.5, 'morning chat')",
            [],
        )
        .unwrap();
        insert_session(&conn, "sess-b", 3, 5);
        insert_entity(&conn, 10, "Alice", "person");
        insert_entity(&conn, 20, "Bob", "person");
        insert_entity(&conn, 30, "tank", "machine");
        insert_relation(&conn, 10, 20, "knows");
        conn.execute(
            "INSERT INTO relations VALUES (20, 30, 'uses', 0.25, 1700000456.0)",
            [],
        )
        .unwrap();
        drop(conn);

        let mut args = vec![
            "--agent-id",
            "e2e-agent",
            "--embedder",
            "minilm",
            "--validate-full",
        ];
        args.extend_from_slice(extra);
        let o = migrate_ok(&db_path, &h5_path, &args);
        assert_eq!(o.summary.float16, expect_float16);
        assert_eq!(o.summary.count, 6);
        assert_eq!(o.summary.active, 5);

        // Read-only view: the whole store, as on disk.
        let source =
            sqlite_reader::read_sqlite(&db_path, false, None, &SchemaConfig::default()).unwrap();
        let ro = HDF5Memory::open_read_only(&h5_path).unwrap();
        assert_eq!(ro.count(), 6);
        assert_eq!(ro.count_active(), 5);
        assert_eq!(ro.config().agent_id, "e2e-agent");
        assert_eq!(ro.config().embedder, "minilm");
        assert_eq!(ro.config().embedding_dim, dim);
        assert_eq!(ro.config().float16, expect_float16);
        for (i, c) in source.chunks.iter().enumerate() {
            let want: Vec<u32> = c
                .embedding
                .iter()
                .map(|&v| if expect_float16 { round_to_f16(v) } else { v }.to_bits())
                .collect();
            let got: Vec<u32> = ro.cache.embeddings[i].iter().map(|v| v.to_bits()).collect();
            assert_eq!(got, want, "record {i}");
        }
        drop(ro);

        // Writable open: a real agent store, searchable, with sessions and KG.
        let mut mem = HDF5Memory::open(&h5_path).unwrap();
        assert_eq!(mem.count(), 6);
        let hits = mem.hybrid_search(&axis_embedding(dim, 9), "allergic peanuts", 0.7, 0.3, 3);
        assert_eq!(hits[0].chunk, texts[3], "{hits:?}");
        let hits = mem.search(
            &axis_embedding(dim, 6),
            "build cache runner",
            &SearchOptions::new(3).with_sources(["api"]),
        );
        assert_eq!(hits[0].chunk, texts[2], "{hits:?}");
        // The deleted source row is a tombstone: never returned.
        let hits = mem.hybrid_search(&axis_embedding(dim, 15), "forgotten memory", 0.5, 0.5, 10);
        assert!(
            hits.iter().all(|h| h.chunk != "a forgotten memory"),
            "{hits:?}"
        );

        assert_eq!(
            mem.get_session_summary("sess-a").unwrap().as_deref(),
            Some("morning chat")
        );
        let s = &mem.sessions().entries[0];
        assert_eq!(
            (s.start_idx, s.end_idx, s.channel.as_str()),
            (0, 2, "discord")
        );
        assert_eq!(s.ts, 1700000123.5 * 1e6, "seconds -> microseconds");
        assert_eq!(mem.sessions().len(), 2);

        let kg = mem.knowledge();
        assert_eq!(kg.entities.len(), 3);
        assert_eq!(kg.relations.len(), 2);
        let name = |id: u64| kg.get_entity(id).unwrap().name.clone();
        let rel: Vec<(String, String, String, f32)> = kg
            .relations
            .iter()
            .map(|r| (name(r.src), r.relation.clone(), name(r.tgt), r.weight))
            .collect();
        assert_eq!(
            rel,
            vec![
                ("Alice".into(), "knows".into(), "Bob".into(), 1.0),
                ("Bob".into(), "uses".into(), "tank".into(), 0.25),
            ]
        );
        assert_eq!(kg.relations[1].ts, 1700000456.0 * 1e6);
        assert_eq!(kg.get_entity(2).unwrap().entity_type, "machine");

        // And it keeps working as a store: a new save survives a reopen.
        mem.save(clawhdf5_agent::MemoryEntry {
            chunk: "saved after migration".into(),
            embedding: axis_embedding(dim, 1),
            source_channel: "api".into(),
            timestamp: 1.0,
            session_id: String::new(),
            tags: String::new(),
        })
        .unwrap();
        drop(mem);
        assert_eq!(HDF5Memory::open(&h5_path).unwrap().count(), 7);
    }

    #[test]
    fn end_to_end_float16_default() {
        end_to_end(&[], true);
    }

    #[test]
    fn end_to_end_f32() {
        end_to_end(&["--f32"], false);
    }

    #[test]
    fn hidden_float16_flag_is_a_no_op() {
        end_to_end(&["--float16"], true);
    }

    #[test]
    fn cli_precision_flags() {
        let base = ["m", "--sqlite", "a.db", "--hdf5", "b.h5"];
        let parse = |extra: &[&str]| {
            let mut args = base.to_vec();
            args.extend_from_slice(extra);
            Cli::try_parse_from(args)
        };
        let c = parse(&[]).unwrap();
        assert!(!c.f32 && !c.float16);
        assert!(parse(&["--f32"]).unwrap().f32);
        assert!(parse(&["--float16"]).is_ok());
        assert!(parse(&["--f32", "--float16"]).is_err());
        // --float16 stays accepted but out of the help text.
        use clap::CommandFactory;
        let help = Cli::command().render_long_help().to_string();
        assert!(help.contains("--f32"));
        assert!(!help.contains("--float16"), "{help}");
    }

    // ---------- Skip deleted rows ----------
    #[test]
    fn test_skip_deleted() {
        let dir = TempDir::new().unwrap();
        let db_path = create_test_db(&dir);
        let h5_path = out_path(&dir, "out.h5");

        let conn = Connection::open(&db_path).unwrap();
        insert_chunk(&conn, 1, "active", &make_embedding(4, 1.0), 0);
        insert_chunk(&conn, 2, "deleted", &make_embedding(4, 2.0), 1);
        insert_chunk(&conn, 3, "also active", &make_embedding(4, 3.0), 0);
        drop(conn);

        let o = migrate_ok(&db_path, &h5_path, &["--skip-deleted"]);
        assert_eq!(o.summary.count, 2);
        assert_eq!(o.summary.active, 2);
    }

    // ---------- Include deleted rows (as tombstones) ----------
    #[test]
    fn test_include_deleted() {
        let dir = TempDir::new().unwrap();
        let db_path = create_test_db(&dir);
        let h5_path = out_path(&dir, "out.h5");

        let conn = Connection::open(&db_path).unwrap();
        insert_chunk(&conn, 1, "active", &make_embedding(4, 1.0), 0);
        insert_chunk(&conn, 2, "deleted", &make_embedding(4, 2.0), 1);
        drop(conn);

        let o = migrate_ok(&db_path, &h5_path, &[]);
        assert_eq!(o.summary.count, 2);
        assert_eq!(o.summary.active, 1);
        let mem = HDF5Memory::open_read_only(&h5_path).unwrap();
        assert_eq!(mem.cache.tombstones, vec![0, 1]);
    }

    // ---------- Auto-detect embedding dimension ----------
    #[test]
    fn test_auto_detect_dim() {
        let dir = TempDir::new().unwrap();
        let db_path = create_test_db(&dir);

        let conn = Connection::open(&db_path).unwrap();
        insert_chunk(&conn, 1, "test", &make_embedding(16, 0.5), 0);
        drop(conn);

        let data =
            sqlite_reader::read_sqlite(&db_path, false, None, &SchemaConfig::default()).unwrap();
        assert_eq!(data.embedding_dim, 16);
    }

    // ---------- Manual embedding dimension ----------
    #[test]
    fn test_manual_dim() {
        let dir = TempDir::new().unwrap();
        let db_path = create_test_db(&dir);
        let h5_path = out_path(&dir, "out.h5");

        let conn = Connection::open(&db_path).unwrap();
        insert_chunk(&conn, 1, "test", &make_embedding(16, 0.5), 0);
        drop(conn);

        let data =
            sqlite_reader::read_sqlite(&db_path, false, Some(8), &SchemaConfig::default()).unwrap();
        assert_eq!(data.embedding_dim, 8);
        // Read at full length, never truncated to the requested dimension...
        assert_eq!(data.chunks[0].embedding.len(), 16);

        // ...so a --embedding-dim that disagrees with the data is an error.
        let err = migrate(&db_path, &h5_path, &["--embedding-dim", "8"])
            .unwrap_err()
            .to_string();
        assert!(
            err.contains("chunk id 1") && err.contains("16 values, expected 8"),
            "{err}"
        );
        assert!(!h5_path.exists(), "nothing written");
        let o = migrate_ok(&db_path, &h5_path, &["--embedding-dim", "16"]);
        assert_eq!(o.summary.embedding_dim, 16);
        let err = migrate(&db_path, &h5_path, &["--embedding-dim", "0"])
            .unwrap_err()
            .to_string();
        assert!(err.contains("at least 1"), "{err}");
    }

    // ---------- A row longer than the first is an error, not truncated ----------
    #[test]
    fn long_embedding_is_rejected() {
        let dir = TempDir::new().unwrap();
        let db_path = create_test_db(&dir);
        let h5_path = out_path(&dir, "out.h5");

        let conn = Connection::open(&db_path).unwrap();
        insert_chunk(&conn, 1, "eight", &make_embedding(8, 0.5), 0);
        insert_chunk(&conn, 2, "twelve", &make_embedding(12, 0.5), 0);
        drop(conn);

        let err = migrate(&db_path, &h5_path, &[]).unwrap_err().to_string();
        assert!(
            err.contains("chunk id 2") && err.contains("12 values, expected 8"),
            "{err}"
        );
        assert!(!h5_path.exists(), "nothing written");
    }

    // ---------- A short first row does not set a small dimension for all ----------
    #[test]
    fn short_first_row_is_rejected_not_imposed() {
        let dir = TempDir::new().unwrap();
        let db_path = create_test_db(&dir);
        let h5_path = out_path(&dir, "out.h5");

        let conn = Connection::open(&db_path).unwrap();
        insert_chunk(&conn, 1, "four", &make_embedding(4, 0.5), 0);
        insert_chunk(&conn, 2, "eight", &make_embedding(8, 0.5), 0);
        insert_chunk(&conn, 3, "eight too", &make_embedding(8, 0.7), 0);
        drop(conn);

        let err = migrate(&db_path, &h5_path, &[]).unwrap_err().to_string();
        assert!(err.contains("chunk id 2"), "{err}");
        let err = migrate(&db_path, &h5_path, &["--embedding-dim", "8"])
            .unwrap_err()
            .to_string();
        assert!(
            err.contains("chunk id 1") && err.contains("4 values, expected 8"),
            "{err}"
        );
        assert!(!h5_path.exists(), "nothing written");
    }

    // ---------- A BLOB that is not whole f32 values is an error ----------
    #[test]
    fn ragged_embedding_blob_is_rejected() {
        let dir = TempDir::new().unwrap();
        let db_path = create_test_db(&dir);
        let h5_path = out_path(&dir, "out.h5");

        let conn = Connection::open(&db_path).unwrap();
        insert_chunk(&conn, 1, "ok", &make_embedding(4, 0.5), 0);
        conn.execute(
            "INSERT INTO memory_chunks (id, chunk, embedding, timestamp, deleted)
             VALUES (2, 'ragged', ?1, 1700000000.0, 0)",
            [vec![0u8; 18]],
        )
        .unwrap();
        drop(conn);

        let err = migrate(&db_path, &h5_path, &[]).unwrap_err().to_string();
        assert!(
            err.contains("chunk id 2") && err.contains("18 bytes"),
            "{err}"
        );
    }

    // ---------- Rows with no embedding are rejected before anything is written ----------
    #[test]
    fn empty_embeddings_are_rejected_and_keep_the_old_store() {
        let dir = TempDir::new().unwrap();
        let db_path = create_test_db(&dir);
        let h5_path = out_path(&dir, "out.h5");

        let good = dir.path().join("good.db");
        std::fs::copy(&db_path, &good).unwrap();
        let conn = Connection::open(&good).unwrap();
        insert_chunk(&conn, 1, "one", &make_embedding(4, 1.0), 0);
        drop(conn);
        migrate_ok(good.to_str().unwrap(), &h5_path, &[]);

        let conn = Connection::open(&db_path).unwrap();
        insert_chunk(&conn, 1, "a", &[], 0);
        insert_chunk(&conn, 2, "b", &[], 0);
        drop(conn);
        for extra in [&[][..], &["--embedding-dim", "4"][..]] {
            let err = migrate(&db_path, &h5_path, extra).unwrap_err().to_string();
            assert!(
                err.contains("chunk id 1") && err.contains("embedding is empty"),
                "{extra:?}: {err}"
            );
        }
        assert_eq!(HDF5Memory::open(&h5_path).unwrap().count(), 1);

        // The same into a path with nothing there: no store is left behind.
        let fresh = out_path(&dir, "fresh.h5");
        assert!(migrate(&db_path, &fresh, &[]).is_err());
        assert!(!fresh.exists());
    }

    // ---------- A failed run leaves the existing store as it was ----------
    #[test]
    fn failed_run_keeps_the_existing_store() {
        let dir = TempDir::new().unwrap();
        let db_path = create_test_db(&dir);
        let h5_path = out_path(&dir, "out.h5");

        let conn = Connection::open(&db_path).unwrap();
        for i in 1..=3 {
            insert_chunk(&conn, i, &format!("r{i}"), &make_embedding(4, i as f32), 0);
        }
        drop(conn);
        migrate_ok(&db_path, &h5_path, &[]);

        let big = dir.path().join("big.db");
        let big = big.to_str().unwrap();
        std::fs::copy(&db_path, big).unwrap();
        let conn = Connection::open(big).unwrap();
        conn.execute("DELETE FROM memory_chunks", []).unwrap();
        insert_chunk(&conn, 1, "huge", &[70000.0, 0.0, 0.0, 0.0], 0);
        drop(conn);

        let err = migrate(big, &h5_path, &[]).unwrap_err().to_string();
        assert!(err.contains("--f32"), "{err}");
        let mem = HDF5Memory::open(&h5_path).unwrap();
        assert_eq!(mem.count(), 3);
        assert_eq!(mem.cache.chunks, ["r1", "r2", "r3"]);
    }

    // ---------- A row with a short embedding is an error, not zero-padded ----------
    #[test]
    fn short_embedding_is_rejected() {
        let dir = TempDir::new().unwrap();
        let db_path = create_test_db(&dir);
        let h5_path = out_path(&dir, "out.h5");

        let conn = Connection::open(&db_path).unwrap();
        insert_chunk(&conn, 1, "full", &make_embedding(8, 0.5), 0);
        insert_chunk(&conn, 2, "short", &make_embedding(5, 0.5), 0);
        drop(conn);

        let err = migrate(&db_path, &h5_path, &[]).unwrap_err().to_string();
        assert!(
            err.contains("chunk id 2") && err.contains("expected 8"),
            "{err}"
        );
    }

    // ---------- Float16: exact half-precision values ----------
    #[test]
    fn test_float16_conversion() {
        let dir = TempDir::new().unwrap();
        let db_path = create_test_db(&dir);
        let h5_path = out_path(&dir, "out.h5");

        let conn = Connection::open(&db_path).unwrap();
        // 0.1 and 1/3 are not representable in half precision.
        let emb = vec![1.0f32, 0.1, -1.0 / 3.0, 3.125];
        insert_chunk(&conn, 1, "test", &emb, 0);
        drop(conn);

        let o = migrate_ok(&db_path, &h5_path, &["--validate-full"]);
        assert!(o.summary.float16);
        let mem = HDF5Memory::open_read_only(&h5_path).unwrap();
        let got = &mem.cache.embeddings[0];
        for (&v, &g) in emb.iter().zip(got) {
            assert_eq!(g.to_bits(), round_to_f16(v).to_bits());
        }
        assert_ne!(got[1], 0.1, "stored at half precision");
    }

    // ---------- Float16 range: an error that points at --f32 ----------
    #[test]
    fn value_beyond_half_range_needs_f32() {
        let dir = TempDir::new().unwrap();
        let db_path = create_test_db(&dir);
        let h5_path = out_path(&dir, "out.h5");

        let conn = Connection::open(&db_path).unwrap();
        insert_chunk(&conn, 7, "huge", &[1.0, 70000.0, 0.0, 0.0], 0);
        drop(conn);

        let err = migrate(&db_path, &h5_path, &[]).unwrap_err().to_string();
        assert!(err.contains("chunk id 7") && err.contains("--f32"), "{err}");
        let o = migrate_ok(&db_path, &h5_path, &["--f32"]);
        assert!(!o.summary.float16);
        let mem = HDF5Memory::open_read_only(&h5_path).unwrap();
        assert_eq!(mem.cache.embeddings[0][1], 70000.0);
    }

    // ---------- Compression produces a smaller valid store ----------
    #[test]
    fn test_compression() {
        let dir = TempDir::new().unwrap();
        let db_path = create_test_db(&dir);
        let h5_compressed = out_path(&dir, "compressed.h5");
        let h5_uncompressed = out_path(&dir, "uncompressed.h5");

        let conn = Connection::open(&db_path).unwrap();
        // Insert enough data so compression can be effective
        for i in 0..100 {
            insert_chunk(&conn, i, &format!("chunk {i}"), &make_embedding(32, 0.0), 0);
        }
        drop(conn);

        let o = migrate_ok(
            &db_path,
            &h5_compressed,
            &["--compression", "--compression-level", "6"],
        );
        assert_eq!(o.summary.count, 100);
        migrate_ok(&db_path, &h5_uncompressed, &[]);

        let mem = HDF5Memory::open_read_only(&h5_compressed).unwrap();
        assert!(mem.config().compression);
        assert_eq!(mem.config().compression_level, 6);
        let sz_c = std::fs::metadata(&h5_compressed).unwrap().len();
        let sz_u = std::fs::metadata(&h5_uncompressed).unwrap().len();
        assert!(
            sz_c < sz_u,
            "Compressed ({sz_c}) should be smaller than uncompressed ({sz_u})"
        );
    }

    // ---------- Dry run doesn't create file ----------
    #[test]
    fn test_dry_run() {
        let dir = TempDir::new().unwrap();
        let db_path = create_test_db(&dir);
        let h5_path = out_path(&dir, "should_not_exist.h5");

        let conn = Connection::open(&db_path).unwrap();
        insert_chunk(&conn, 1, "test", &make_embedding(4, 1.0), 0);
        drop(conn);

        assert!(
            migrate(&db_path, &h5_path, &["--dry-run"])
                .unwrap()
                .is_none()
        );
        assert!(!h5_path.exists());
    }

    // ---------- Empty database migration ----------
    #[test]
    fn test_empty_db() {
        let dir = TempDir::new().unwrap();
        let db_path = create_test_db(&dir);
        let h5_path = out_path(&dir, "out.h5");

        let data =
            sqlite_reader::read_sqlite(&db_path, false, None, &SchemaConfig::default()).unwrap();
        assert_eq!(data.chunks.len(), 0);
        assert_eq!(data.sessions.len(), 0);
        assert_eq!(data.entities.len(), 0);
        assert_eq!(data.relations.len(), 0);

        // No record to detect the dimension from: a store with dimension 0
        // could never take a record, so --embedding-dim is required.
        let err = migrate(&db_path, &h5_path, &[]).unwrap_err().to_string();
        assert!(err.contains("--embedding-dim"), "{err}");
        assert!(!h5_path.exists());

        let o = migrate_ok(&db_path, &h5_path, &["--embedding-dim", "8"]);
        assert_eq!(o.summary.count, 0);
        assert_eq!(o.summary.embedding_dim, 8);
        assert!(!o.summary.search_checked);
        assert_eq!(HDF5Memory::open(&h5_path).unwrap().count(), 0);

        // ...and a later incremental run can add records to it.
        let conn = Connection::open(&db_path).unwrap();
        insert_chunk(&conn, 1, "later", &make_embedding(8, 0.5), 0);
        drop(conn);
        let o = migrate_ok(&db_path, &h5_path, &["--incremental"]);
        assert_eq!((o.summary.count, o.summary.embedding_dim), (1, 8));
        assert!(o.summary.search_checked);
    }

    // ---------- A source with only sessions/graph still needs a dimension ----------
    #[test]
    fn graph_only_source_needs_embedding_dim() {
        let dir = TempDir::new().unwrap();
        let db_path = create_test_db(&dir);
        let h5_path = out_path(&dir, "out.h5");

        let conn = Connection::open(&db_path).unwrap();
        insert_entity(&conn, 1, "Alice", "person");
        insert_session(&conn, "s1", 0, 0);
        drop(conn);
        let err = migrate(&db_path, &h5_path, &[]).unwrap_err().to_string();
        assert!(err.contains("--embedding-dim"), "{err}");
        let o = migrate_ok(&db_path, &h5_path, &["--embedding-dim", "4"]);
        assert_eq!((o.summary.entities, o.summary.sessions), (1, 1));
        assert_eq!(o.summary.embedding_dim, 4);
    }

    // ---------- Large migration (1000 entries) ----------
    #[test]
    fn test_large_migration() {
        let dir = TempDir::new().unwrap();
        let db_path = create_test_db(&dir);
        let h5_path = out_path(&dir, "out.h5");

        let conn = Connection::open(&db_path).unwrap();
        for i in 0..1000 {
            insert_chunk(
                &conn,
                i,
                &format!("chunk number {i} with some text"),
                &make_embedding(64, i as f32 * 0.01),
                0,
            );
        }
        drop(conn);

        let o = migrate_ok(&db_path, &h5_path, &[]);
        assert_eq!(o.summary.count, 1000);
        assert!(o.summary.search_checked);
        // Sampled, not every row, without --validate-full.
        assert!(o.summary.rows_checked < 1000);
        let o = migrate_ok(&db_path, &h5_path, &["--validate-full"]);
        assert_eq!(o.summary.rows_checked, 1000);
    }

    // ---------- Session migration ----------
    #[test]
    fn test_session_migration() {
        let dir = TempDir::new().unwrap();
        let db_path = create_test_db(&dir);
        let h5_path = out_path(&dir, "out.h5");

        let conn = Connection::open(&db_path).unwrap();
        insert_session(&conn, "session-alpha", 0, 10);
        insert_session(&conn, "session-beta", 11, 20);
        insert_session(&conn, "session-gamma", 21, 30);
        drop(conn);

        let o = migrate_ok(&db_path, &h5_path, &["--embedding-dim", "8"]);
        assert_eq!(o.summary.sessions, 3);
        let mem = HDF5Memory::open(&h5_path).unwrap();
        let ids: Vec<&str> = mem
            .sessions()
            .entries
            .iter()
            .map(|e| e.id.as_str())
            .collect();
        assert_eq!(ids, ["session-alpha", "session-beta", "session-gamma"]);
        assert_eq!(
            mem.get_session_summary("session-beta").unwrap().as_deref(),
            Some("test summary")
        );
    }

    // ---------- Knowledge graph (entities + relations) ----------
    #[test]
    fn test_knowledge_graph_migration() {
        let dir = TempDir::new().unwrap();
        let db_path = create_test_db(&dir);
        let h5_path = out_path(&dir, "out.h5");

        let conn = Connection::open(&db_path).unwrap();
        insert_entity(&conn, 1, "Alice", "person");
        insert_entity(&conn, 2, "Bob", "person");
        insert_entity(&conn, 3, "Rust", "language");
        insert_relation(&conn, 1, 2, "knows");
        insert_relation(&conn, 1, 3, "uses");
        insert_relation(&conn, 2, 3, "uses");
        drop(conn);

        let o = migrate_ok(&db_path, &h5_path, &["--embedding-dim", "8"]);
        assert_eq!(o.summary.entities, 3);
        assert_eq!(o.summary.relations, 3);
        let mem = HDF5Memory::open(&h5_path).unwrap();
        let kg = mem.knowledge();
        let alice = o.migration.entity_ids[&1];
        let targets: Vec<&str> = kg
            .get_relations_from(alice)
            .iter()
            .map(|r| kg.get_entity(r.tgt).unwrap().name.as_str())
            .collect();
        assert_eq!(targets, ["Bob", "Rust"]);
    }

    // ---------- A relation to an unknown entity is skipped and reported ----------
    #[test]
    fn dangling_relation_is_reported() {
        let dir = TempDir::new().unwrap();
        let db_path = create_test_db(&dir);
        let h5_path = out_path(&dir, "out.h5");

        let conn = Connection::open(&db_path).unwrap();
        insert_entity(&conn, 1, "Alice", "person");
        insert_relation(&conn, 1, 1, "self");
        // A database that did not enforce its foreign keys.
        conn.execute_batch("PRAGMA foreign_keys = OFF").unwrap();
        insert_relation(&conn, 1, 42, "knows");
        drop(conn);

        let o = migrate_ok(&db_path, &h5_path, &["--embedding-dim", "8"]);
        assert_eq!(o.summary.relations, 1);
        assert_eq!(o.migration.dangling_relations, vec![1]);
    }

    // ---------- Validation catches a record count mismatch ----------
    #[test]
    fn test_validation_catches_count_mismatch() {
        let dir = TempDir::new().unwrap();
        let db_path = create_test_db(&dir);
        let h5_path = out_path(&dir, "out.h5");

        let conn = Connection::open(&db_path).unwrap();
        insert_chunk(&conn, 1, "test", &make_embedding(4, 1.0), 0);
        drop(conn);

        let (data, mut m) = write(&db_path, &h5_path, false);
        m.store_count += 1;
        let result = validate::validate_store(&h5_path, &data, &m, false);
        assert!(result.unwrap_err().to_string().contains("count mismatch"));
    }

    // ---------- Metadata is stored in the agent config ----------
    #[test]
    fn test_metadata_attributes() {
        let dir = TempDir::new().unwrap();
        let db_path = create_test_db(&dir);
        let h5_path = out_path(&dir, "out.h5");

        let conn = Connection::open(&db_path).unwrap();
        insert_chunk(&conn, 1, "test", &make_embedding(8, 1.0), 0);
        drop(conn);

        migrate_ok(
            &db_path,
            &h5_path,
            &["--agent-id", "my-agent-42", "--embedder", "openai-ada"],
        );
        let mem = HDF5Memory::open_read_only(&h5_path).unwrap();
        assert_eq!(mem.config().agent_id, "my-agent-42");
        assert_eq!(mem.config().embedder, "openai-ada");
        assert_eq!(mem.config().embedding_dim, 8);
    }

    // ---------- f32 embedding values roundtrip exactly ----------
    #[test]
    fn test_embedding_roundtrip() {
        let dir = TempDir::new().unwrap();
        let db_path = create_test_db(&dir);
        let h5_path = out_path(&dir, "out.h5");

        let conn = Connection::open(&db_path).unwrap();
        let emb = vec![0.1, 0.2, 0.3, 0.4];
        insert_chunk(&conn, 1, "test", &emb, 0);
        drop(conn);

        migrate_ok(&db_path, &h5_path, &["--f32"]);
        let mem = HDF5Memory::open_read_only(&h5_path).unwrap();
        assert_eq!(&mem.cache.embeddings[0], &emb[..]);
    }

    // ---------- Full combined migration ----------
    #[test]
    fn test_full_combined_migration() {
        let dir = TempDir::new().unwrap();
        let db_path = create_test_db(&dir);
        let h5_path = out_path(&dir, "out.h5");

        let conn = Connection::open(&db_path).unwrap();
        for i in 0..5 {
            insert_chunk(
                &conn,
                i,
                &format!("chunk {i}"),
                &make_embedding(16, i as f32),
                if i == 3 { 1 } else { 0 },
            );
        }
        insert_session(&conn, "s1", 0, 2);
        insert_session(&conn, "s2", 3, 4);
        insert_entity(&conn, 1, "Alice", "person");
        insert_entity(&conn, 2, "Bob", "person");
        insert_relation(&conn, 1, 2, "knows");
        drop(conn);

        let o = migrate_ok(
            &db_path,
            &h5_path,
            &[
                "--agent-id",
                "combined-test",
                "--embedder",
                "test-embedder",
                "--skip-deleted",
                "--compression",
            ],
        );
        assert_eq!(o.summary.count, 4); // chunk 3 is deleted
        assert_eq!(o.summary.sessions, 2);
        assert_eq!(o.summary.entities, 2);
        assert_eq!(o.summary.relations, 1);
        assert_eq!(o.summary.embedding_dim, 16);
    }

    // ---------- Validation catches a session mismatch ----------
    #[test]
    fn test_validation_session_mismatch() {
        let dir = TempDir::new().unwrap();
        let db_path = create_test_db(&dir);
        let h5_path = out_path(&dir, "out.h5");

        let conn = Connection::open(&db_path).unwrap();
        insert_session(&conn, "s1", 0, 10);
        insert_chunk(&conn, 1, "one", &make_embedding(8, 1.0), 0);
        drop(conn);

        let (mut data, m) = write(&db_path, &h5_path, false);
        validate::validate_store(&h5_path, &data, &m, false).unwrap();
        data.sessions[0].summary = "DIFFERENT".into();
        let result = validate::validate_store(&h5_path, &data, &m, false);
        assert!(result.unwrap_err().to_string().contains("session"));
    }

    // ---------- Content validation catches corrupt embeddings ----------
    #[test]
    fn test_content_validation_catches_embedding_corruption() {
        let dir = TempDir::new().unwrap();
        let db_path = create_test_db(&dir);
        let h5_path = out_path(&dir, "out.h5");

        let conn = Connection::open(&db_path).unwrap();
        insert_chunk(&conn, 1, "hello", &make_embedding(8, 1.0), 0);
        drop(conn);

        for f32 in [false, true] {
            let (mut data, m) = write(&db_path, &h5_path, f32);
            validate::validate_store(&h5_path, &data, &m, true).unwrap();
            // Exact, not within a tolerance: one ulp is a mismatch.
            let v = &mut data.chunks[0].embedding[3];
            *v = if f32 {
                f32::from_bits(v.to_bits() + 1)
            } else {
                // The next half-precision value up.
                clawhdf5_format::float16::f16_bits_to_f32(
                    clawhdf5_format::float16::f32_to_f16_bits(*v) + 1,
                )
            };
            let result = validate::validate_store(&h5_path, &data, &m, true);
            assert!(
                result.unwrap_err().to_string().contains("embedding"),
                "f32={f32}"
            );
        }
    }

    // ---------- Configurable schema: custom table names ----------
    #[test]
    fn test_configurable_table_names() {
        let dir = TempDir::new().unwrap();
        let db_path = dir.path().join("custom.db");
        let path_str = db_path.to_str().unwrap().to_string();
        let conn = Connection::open(&path_str).unwrap();
        // Chunks live in a differently-named table; the others use defaults.
        conn.execute_batch(
            "CREATE TABLE my_chunks (
                id INTEGER PRIMARY KEY, chunk TEXT, embedding BLOB,
                source_channel TEXT, timestamp REAL, session_id TEXT, tags TEXT, deleted INTEGER
            );
            CREATE TABLE sessions (id TEXT, start_idx INTEGER, end_idx INTEGER, channel TEXT, timestamp REAL, summary TEXT);
            CREATE TABLE entities (id INTEGER, name TEXT, type TEXT, embedding_idx INTEGER);
            CREATE TABLE relations (src INTEGER, tgt INTEGER, relation TEXT, weight REAL, timestamp REAL);",
        )
        .unwrap();
        let blob: Vec<u8> = make_embedding(4, 1.0)
            .iter()
            .flat_map(|v| v.to_le_bytes())
            .collect();
        conn.execute(
            "INSERT INTO my_chunks VALUES (1, 'hi', ?1, 'api', 1.0, 's', '', 0)",
            rusqlite::params![blob],
        )
        .unwrap();
        drop(conn);

        let mut schema = SchemaConfig::default();
        schema.chunks.table = "my_chunks".into();
        let data = sqlite_reader::read_sqlite(&path_str, false, None, &schema).unwrap();
        assert_eq!(data.chunks.len(), 1);
        assert_eq!(data.chunks[0].chunk, "hi");
        assert_eq!(data.embedding_dim, 4);

        // Counts pass should also honor the custom table name.
        let counts = sqlite_reader::read_counts(&path_str, false, &schema).unwrap();
        assert_eq!(counts.chunks, 1);

        // And the CLI flag reaches the reader.
        let h5_path = out_path(&dir, "out.h5");
        let o = migrate_ok(
            &path_str,
            &h5_path,
            &["--chunks-table", "my_chunks", "--verbose"],
        );
        assert_eq!(o.summary.count, 1);
    }

    // ---------- Incremental migration appends only new rows ----------
    #[test]
    fn test_incremental_migration() {
        let dir = TempDir::new().unwrap();
        let db_path = create_test_db(&dir);
        let h5_path = out_path(&dir, "out.h5");

        // First migration: 2 chunks, a session and an edge.
        let conn = Connection::open(&db_path).unwrap();
        insert_chunk(&conn, 1, "one", &make_embedding(4, 1.0), 0);
        insert_chunk(&conn, 2, "two", &make_embedding(4, 2.0), 0);
        insert_session(&conn, "s1", 0, 1);
        insert_entity(&conn, 1, "Alice", "person");
        insert_entity(&conn, 2, "Bob", "person");
        insert_relation(&conn, 1, 2, "knows");
        drop(conn);
        let o = migrate_ok(&db_path, &h5_path, &["--incremental", "--f32"]);
        assert!(!o.migration.appended_to_existing);
        assert_eq!(o.summary.count, 2);

        // The agent uses the store in between.
        let mut mem = HDF5Memory::open(&h5_path).unwrap();
        mem.add_entity("Carol", "person", -1).unwrap();
        drop(mem);

        // Add rows, then migrate incrementally.
        let conn = Connection::open(&db_path).unwrap();
        insert_chunk(&conn, 3, "three", &make_embedding(4, 3.0), 0);
        insert_chunk(&conn, 4, "four", &make_embedding(4, 4.0), 1);
        insert_session(&conn, "s2", 2, 3);
        insert_entity(&conn, 3, "Rust", "language");
        insert_relation(&conn, 1, 3, "uses");
        drop(conn);

        // --f32 is irrelevant here: the store keeps its precision anyway.
        let o = migrate_ok(&db_path, &h5_path, &["--incremental", "--validate-full"]);
        let m = &o.migration;
        assert!(m.appended_to_existing);
        assert!(!o.summary.float16, "an existing store keeps its precision");
        assert_eq!((m.records.len(), m.chunks_present), (2, 2));
        assert_eq!((m.sessions.len(), m.sessions_present), (1, 1));
        assert_eq!((m.entities.len(), m.entities_present), (1, 2));
        assert_eq!((m.relations.len(), m.relations_present), (1, 1));
        assert_eq!(o.summary.count, 4);
        assert_eq!(o.summary.active, 3);
        assert_eq!(o.summary.sessions, 2);
        assert_eq!(o.summary.entities, 4); // Alice, Bob, Carol, Rust
        assert_eq!(o.summary.relations, 2);

        let mem = HDF5Memory::open_read_only(&h5_path).unwrap();
        assert_eq!(mem.cache.chunks, ["one", "two", "three", "four"]);
        assert_eq!(mem.cache.tombstones, [0, 0, 0, 1]);
        drop(mem);

        // Nothing new: nothing written.
        let o = migrate_ok(&db_path, &h5_path, &["--incremental"]);
        assert!(o.migration.records.is_empty() && o.migration.relations.is_empty());
        assert_eq!(o.summary.count, 4);
    }

    // ---------- --incremental: a source of another dimension is an error ----------
    #[test]
    fn incremental_rejects_a_different_dimension() {
        let dir = TempDir::new().unwrap();
        let db_path = create_test_db(&dir);
        let h5_path = out_path(&dir, "out.h5");

        let conn = Connection::open(&db_path).unwrap();
        insert_chunk(&conn, 1, "one", &make_embedding(8, 1.0), 0);
        drop(conn);
        migrate_ok(&db_path, &h5_path, &[]);

        let other = dir.path().join("other.db");
        let other = other.to_str().unwrap();
        std::fs::copy(&db_path, other).unwrap();
        let conn = Connection::open(other).unwrap();
        conn.execute("DELETE FROM memory_chunks", []).unwrap();
        insert_chunk(&conn, 1, "sixteen", &make_embedding(16, 1.0), 0);
        drop(conn);

        for extra in [
            &["--incremental"][..],
            &["--incremental", "--embedding-dim", "8"],
        ] {
            let err = migrate(other, &h5_path, extra).unwrap_err().to_string();
            assert!(err.contains("embedding"), "{extra:?}: {err}");
        }
        let err = migrate(other, &h5_path, &["--incremental"])
            .unwrap_err()
            .to_string();
        assert!(err.contains("has embedding_dim 8, the source 16"), "{err}");
        let mem = HDF5Memory::open(&h5_path).unwrap();
        assert_eq!(mem.count(), 1);
        assert_eq!(mem.cache.embeddings[0].len(), 8);
    }

    // ---------- --incremental carries over changes to the deleted flag ----------
    fn deleted_flag_is_reconciled(skip_deleted: bool) {
        let dir = TempDir::new().unwrap();
        let db_path = create_test_db(&dir);
        let h5_path = out_path(&dir, "out.h5");
        let skip: &[&str] = if skip_deleted {
            &["--skip-deleted"]
        } else {
            &[]
        };
        let with = |more: &[&'static str]| -> Vec<&str> {
            let mut v = skip.to_vec();
            v.extend_from_slice(more);
            v
        };

        let conn = Connection::open(&db_path).unwrap();
        for i in 1..=5 {
            insert_chunk(
                &conn,
                i,
                &format!("r{i}"),
                &axis_embedding(8, i as usize),
                0,
            );
        }
        insert_chunk(&conn, 6, "r6", &axis_embedding(8, 6), 1);
        drop(conn);
        migrate_ok(&db_path, &h5_path, &with(&["--incremental"]));
        let before = if skip_deleted { 5 } else { 6 };
        assert_eq!(HDF5Memory::open(&h5_path).unwrap().count(), before);

        // Row 5 deleted and row 6 restored in the source since then.
        let conn = Connection::open(&db_path).unwrap();
        conn.execute("UPDATE memory_chunks SET deleted = 1 WHERE id = 5", [])
            .unwrap();
        conn.execute("UPDATE memory_chunks SET deleted = 0 WHERE id = 6", [])
            .unwrap();
        drop(conn);
        let o = migrate_ok(
            &db_path,
            &h5_path,
            &with(&["--incremental", "--validate-full"]),
        );
        let m = &o.migration;
        assert_eq!(m.deleted_in_store.len(), 1);
        assert_eq!(m.restored, usize::from(!skip_deleted));
        assert_eq!(m.records.len(), 1, "r6 written (again)");
        assert_eq!(m.chunks_present, 4);

        let mut mem = HDF5Memory::open_read_only(&h5_path).unwrap();
        let active: Vec<&str> = (0..mem.count())
            .filter(|&i| mem.cache.tombstones[i] == 0)
            .map(|i| mem.cache.chunks[i].as_str())
            .collect();
        assert_eq!(active, ["r1", "r2", "r3", "r4", "r6"]);
        assert_eq!(o.summary.active, 5);
        // r5 is no longer found by search; r6 is.
        let hits = mem.search(&axis_embedding(8, 5), "r5", &SearchOptions::new(10));
        assert!(hits.iter().all(|h| h.chunk != "r5"), "{hits:?}");
        let hits = mem.search(&axis_embedding(8, 6), "r6", &SearchOptions::new(10));
        assert!(hits.iter().any(|h| h.chunk == "r6"), "{hits:?}");
        drop(mem);

        // Idempotent: a second run changes nothing.
        let o = migrate_ok(&db_path, &h5_path, &with(&["--incremental"]));
        let m = &o.migration;
        assert!(m.records.is_empty() && m.deleted_in_store.is_empty() && m.restored == 0);
        assert_eq!(o.summary.active, 5);
    }

    #[test]
    fn incremental_follows_source_deletes_and_restores() {
        deleted_flag_is_reconciled(false);
    }

    #[test]
    fn incremental_skip_deleted_still_follows_source_deletes() {
        deleted_flag_is_reconciled(true);
    }

    // ---------- Without --incremental an existing store is replaced ----------
    #[test]
    fn rerun_without_incremental_replaces_store() {
        let dir = TempDir::new().unwrap();
        let db_path = create_test_db(&dir);
        let h5_path = out_path(&dir, "out.h5");

        let conn = Connection::open(&db_path).unwrap();
        insert_chunk(&conn, 1, "one", &make_embedding(4, 1.0), 0);
        drop(conn);
        migrate_ok(&db_path, &h5_path, &[]);
        let o = migrate_ok(&db_path, &h5_path, &[]);
        assert_eq!(o.summary.count, 1);
        assert_eq!(HDF5Memory::open(&h5_path).unwrap().count(), 1);
    }

    // ---------- The store is locked while open for writing ----------
    #[test]
    fn migrating_into_a_store_in_use_fails() {
        let dir = TempDir::new().unwrap();
        let db_path = create_test_db(&dir);
        let h5_path = out_path(&dir, "out.h5");

        let conn = Connection::open(&db_path).unwrap();
        insert_chunk(&conn, 1, "one", &make_embedding(4, 1.0), 0);
        drop(conn);
        migrate_ok(&db_path, &h5_path, &[]);
        let _writer = HDF5Memory::open(&h5_path).unwrap();
        for extra in [&[][..], &["--incremental"][..]] {
            let err = migrate(&db_path, &h5_path, extra).unwrap_err().to_string();
            assert!(err.contains("locked"), "{extra:?}: {err}");
        }
    }

    // ---------- h5py can open a migrated store ----------
    fn python() -> String {
        std::env::var("CLAWHDF5_PYTHON").unwrap_or_else(|_| "python3".to_string())
    }

    #[test]
    fn h5py_opens_migrated_store() {
        let has_h5py = std::process::Command::new(python())
            .args(["-c", "import h5py"])
            .output()
            .map(|o| o.status.success())
            .unwrap_or(false);
        if !has_h5py {
            assert!(
                std::env::var("CLAWHDF5_REQUIRE_INTEROP").as_deref() != Ok("1"),
                "CLAWHDF5_REQUIRE_INTEROP=1 but {} with h5py is not available",
                python()
            );
            eprintln!("SKIP: {} with h5py not available", python());
            return;
        }

        let dir = TempDir::new().unwrap();
        let db_path = create_test_db(&dir);
        let conn = Connection::open(&db_path).unwrap();
        for i in 0..6 {
            insert_chunk(
                &conn,
                i,
                &format!("memory {i}"),
                &axis_embedding(8, i as usize),
                0,
            );
        }
        insert_session(&conn, "s1", 0, 5);
        insert_entity(&conn, 1, "Alice", "person");
        insert_entity(&conn, 2, "Bob", "person");
        insert_relation(&conn, 1, 2, "knows");
        drop(conn);

        for (extra, dtype) in [(&[][..], "float16"), (&["--f32"][..], "float32")] {
            let h5_path = out_path(&dir, &format!("{dtype}.h5"));
            migrate_ok(&db_path, &h5_path, extra);
            let bits = (0..6)
                .flat_map(|i| axis_embedding(8, i))
                .map(|v| v.to_bits().to_string())
                .collect::<Vec<_>>()
                .join(",");
            let script = format!(
                r#"
import h5py, numpy as np
with h5py.File(r"{path}", "r") as f:
    names = []
    f.visititems(lambda n, o: names.append(n) if isinstance(o, h5py.Dataset) else None)
    for n in names:
        f[n][()]  # every dataset must decode
    assert f["meta"].attrs["schema_version"] in (b"1.0", "1.0")
    e = f["memory/embeddings"]
    assert e.dtype == np.{dtype}, e.dtype
    ref = np.array([{bits}], dtype=np.uint32).view(np.float32).astype(np.{dtype}).reshape(6, 8)
    assert (e[()] == ref).all()
    assert len(f["memory/chunks"]) == 6
    assert len(f["sessions/ids"]) == 1
    assert len(f["knowledge_graph/entity_ids"]) == 2
    assert len(f["knowledge_graph/relation_srcs"]) == 1
print(len(names))
"#,
                path = h5_path.display()
            );
            let out = std::process::Command::new(python())
                .args(["-c", &script])
                .output()
                .unwrap();
            assert!(
                out.status.success(),
                "{dtype}: {}",
                String::from_utf8_lossy(&out.stderr)
            );
        }
    }
}