diff --git a/crates/clawhdf5-agent/src/lib.rs b/crates/clawhdf5-agent/src/lib.rs index b74507f..62575fa 100644 --- a/crates/clawhdf5-agent/src/lib.rs +++ b/crates/clawhdf5-agent/src/lib.rs @@ -73,7 +73,7 @@ pub use ephemeral::{EphemeralEntry, EphemeralStats}; use knowledge::KnowledgeCache; use memory_strategy::{Exchange, MemoryStrategy, StrategyOutput}; pub use search::SearchOptions; -use session::SessionCache; +pub use session::{SessionCache, SessionEntry}; // --- Error type --- @@ -1028,6 +1028,18 @@ impl HDF5Memory { &self.config } + /// The sessions recorded in this store. + pub fn sessions(&self) -> &SessionCache { + &self.sessions + } + + /// Mutable access to the sessions, e.g. to add many at once. Changes + /// reach the disk at the next checkpoint (any flushing call, such as + /// [`HDF5Memory::flush_wal`] or `save_batch`), not immediately. + pub fn sessions_mut(&mut self) -> &mut SessionCache { + &mut self.sessions + } + /// Get a reference to the knowledge cache. pub fn knowledge(&self) -> &KnowledgeCache { &self.knowledge @@ -1401,6 +1413,35 @@ impl HDF5Memory { } impl HDF5Memory { + /// Delete many records with a single checkpoint, where + /// [`AgentMemory::delete`] checkpoints once per record. + /// + /// All or nothing: if any id is out of range or already deleted (or + /// repeated), nothing is deleted and `MemoryError::NotFound` is returned. + /// Unlike `delete`, this never auto-compacts, so the records stay in the + /// store as tombstones (their indices unchanged) until [`AgentMemory::compact`] + /// is called — importers use it to carry over records that were already + /// deleted in the source. + pub fn delete_batch(&mut self, ids: &[usize]) -> Result<()> { + let mut seen = std::collections::HashSet::with_capacity(ids.len()); + for &id in ids { + if self.cache.tombstones.get(id).copied() != Some(0) || !seen.insert(id) { + return Err(MemoryError::NotFound(format!( + "entry {id} not found or already deleted" + ))); + } + } + if ids.is_empty() { + return Ok(()); + } + for &id in ids { + self.cache.mark_deleted(id); + self.hnsw_on_delete(id); + self.bm25_on_delete(id); + } + self.flush() + } + pub fn tick_session(&mut self) -> Result<()> { let d = self.config.decay_factor; for w in self.cache.activation_weights.iter_mut() { @@ -1599,6 +1640,79 @@ mod tests { } } + #[test] + fn delete_batch_tombstones_without_compacting() { + let dir = TempDir::new().unwrap(); + let path = dir.path().join("test.h5"); + let mut mem = HDF5Memory::create(make_config(&dir)).unwrap(); + mem.save_batch( + (0..4) + .map(|i| make_entry(&format!("record {i}"), &[i as f32, 1.0, 0.0, 0.0])) + .collect(), + ) + .unwrap(); + // 3 of 4 is far past compact_threshold (0.3): delete() would compact. + mem.delete_batch(&[0, 1, 3]).unwrap(); + assert_eq!(mem.count(), 4); + assert_eq!(mem.count_active(), 1); + drop(mem); + + let mut mem = HDF5Memory::open(&path).unwrap(); + assert_eq!(mem.cache.tombstones, vec![1, 1, 0, 1]); + let hits = mem.hybrid_search(&[0.0, 1.0, 0.0, 0.0], "record", 0.5, 0.5, 10); + assert!( + hits.iter().all(|r| r.index == 2), + "tombstoned record returned" + ); + } + + #[test] + fn delete_batch_is_all_or_nothing() { + let dir = TempDir::new().unwrap(); + let mut mem = HDF5Memory::create(make_config(&dir)).unwrap(); + mem.save_batch(vec![ + make_entry("a", &[1.0, 0.0, 0.0, 0.0]), + make_entry("b", &[0.0, 1.0, 0.0, 0.0]), + ]) + .unwrap(); + for bad in [&[0, 5][..], &[1, 1][..]] { + assert!(matches!( + mem.delete_batch(bad), + Err(MemoryError::NotFound(_)) + )); + assert_eq!(mem.count_active(), 2, "{bad:?} deleted something"); + } + mem.delete_batch(&[]).unwrap(); + assert_eq!(mem.count_active(), 2); + } + + #[test] + fn sessions_mut_add_at_keeps_timestamp_across_reopen() { + let dir = TempDir::new().unwrap(); + let path = dir.path().join("test.h5"); + let mut mem = HDF5Memory::create(make_config(&dir)).unwrap(); + mem.sessions_mut() + .add_at("s-old", 2, 7, "discord", "old summary", 1.7e15); + mem.flush_wal().unwrap(); + drop(mem); + + let mem = HDF5Memory::open_read_only(&path).unwrap(); + let s = mem.sessions(); + assert_eq!(s.len(), 1); + let e = &s.entries[0]; + assert_eq!( + ( + e.id.as_str(), + e.start_idx, + e.end_idx, + e.channel.as_str(), + e.ts + ), + ("s-old", 2, 7, "discord", 1.7e15) + ); + assert_eq!(s.summaries[0], "old summary"); + } + #[test] fn create_new_file() { let dir = TempDir::new().unwrap(); diff --git a/crates/clawhdf5-agent/src/session.rs b/crates/clawhdf5-agent/src/session.rs index 9be1f91..ebe817f 100644 --- a/crates/clawhdf5-agent/src/session.rs +++ b/crates/clawhdf5-agent/src/session.rs @@ -33,7 +33,7 @@ impl SessionCache { self.entries.is_empty() } - /// Add a new session with its summary. + /// Add a new session with its summary, timestamped now. pub fn add( &mut self, id: &str, @@ -47,6 +47,21 @@ impl SessionCache { .unwrap_or_default() .as_secs_f64() * 1_000_000.0; // microseconds + self.add_at(id, start_idx, end_idx, channel, summary, ts); + } + + /// Add a session with an explicit timestamp (Unix **microseconds**, the + /// unit [`SessionEntry::ts`] uses) — for importers carrying sessions over + /// from another store, whose original time should be kept. + pub fn add_at( + &mut self, + id: &str, + start_idx: usize, + end_idx: usize, + channel: &str, + summary: &str, + ts: f64, + ) { self.entries.push(SessionEntry { id: id.to_string(), start_idx: start_idx as u64, diff --git a/crates/clawhdf5-migrate/Cargo.toml b/crates/clawhdf5-migrate/Cargo.toml index a518a22..5bd7459 100644 --- a/crates/clawhdf5-migrate/Cargo.toml +++ b/crates/clawhdf5-migrate/Cargo.toml @@ -3,7 +3,7 @@ name = "clawhdf5-migrate" version = "2.7.0" edition = "2024" rust-version.workspace = true -description = "CLI to migrate SQLite agent memory databases to HDF5 format" +description = "CLI to migrate SQLite agent memory databases to clawhdf5-agent stores" license = "MIT" repository = "https://git.redclaw.dev/quantumclaw/clawhdf5" readme = "README.md" @@ -17,10 +17,8 @@ path = "src/main.rs" [dependencies] clawhdf5-agent = { path = "../clawhdf5-agent", version = "2.7.0" } clawhdf5-format = { path = "../clawhdf5-format", version = "2.7.0" } -clawhdf5 = { path = "../clawhdf5", version = "2.7.0" } rusqlite = { version = "0.31", features = ["bundled"] } clap = { version = "4", features = ["derive"] } -half = { workspace = true } [dev-dependencies] tempfile = { workspace = true } diff --git a/crates/clawhdf5-migrate/README.md b/crates/clawhdf5-migrate/README.md index acbb0b2..54b74a6 100644 --- a/crates/clawhdf5-migrate/README.md +++ b/crates/clawhdf5-migrate/README.md @@ -3,9 +3,12 @@ [![crates.io](https://img.shields.io/crates/v/clawhdf5-migrate.svg)](https://crates.io/crates/clawhdf5-migrate) [![docs.rs](https://img.shields.io/docsrs/clawhdf5-migrate)](https://docs.rs/clawhdf5-migrate) -CLI tool to migrate SQLite agent memory databases to HDF5 format. +CLI tool to migrate SQLite agent memory databases (the ZeroClaw layout) to a +[clawhdf5-agent](https://crates.io/crates/clawhdf5-agent) store. -Converts existing SQLite-based agent memory stores (embeddings, text chunks, metadata) into the HDF5 format used by [clawhdf5-agent](https://crates.io/crates/clawhdf5-agent). +The output is written through `clawhdf5-agent`'s own API, so it opens with +`HDF5Memory::open` and is searchable immediately: memory records, sessions and +the knowledge graph (entities and relations) are carried over. ## Installation @@ -16,9 +19,19 @@ cargo install clawhdf5-migrate ## Usage ```bash -clawhdf5-migrate --input agent.db --output agent.h5 +clawhdf5-migrate --sqlite agent.db --hdf5 agent.h5 --agent-id my-agent ``` +Embeddings are stored as float16 (the library default for new stores); pass +`--f32` for full precision. Every embedding must have the same dimension +(the first row's, or `--embedding-dim`, which a source with no memory records +requires); rows are never truncated, and the whole source is checked before an +existing output store is replaced. `--incremental` adds only new rows to an +existing store of the same dimension and carries over changes to rows' +deleted flags, `--skip-deleted` leaves out tombstoned rows, and `--dry-run` +only counts. +See `clawhdf5-migrate --help` for every option. + ## License MIT diff --git a/crates/clawhdf5-migrate/src/hdf5_reader.rs b/crates/clawhdf5-migrate/src/hdf5_reader.rs deleted file mode 100644 index 00c8956..0000000 --- a/crates/clawhdf5-migrate/src/hdf5_reader.rs +++ /dev/null @@ -1,163 +0,0 @@ -//! Read a migration HDF5 file back into the in-memory data model. -//! -//! Used to verify migrated content (real validation) and to merge new rows into -//! an existing output (incremental migration). Mirrors the layout produced by -//! [`crate::hdf5_writer`]. - -use clawhdf5::reader::{File, Group}; -use clawhdf5_format::type_builders::AttrValue; - -use crate::sqlite_reader::{Entity, MemoryChunk, Relation, Session, SqliteData}; - -type BoxErr = Box; - -fn read_strings(group: &Group<'_>, name: &str) -> Result, BoxErr> { - Ok(group.dataset(name)?.read_string()?) -} - -fn read_i64s(group: &Group<'_>, name: &str) -> Result, BoxErr> { - Ok(group.dataset(name)?.read_i64()?) -} - -fn read_f64s(group: &Group<'_>, name: &str) -> Result, BoxErr> { - Ok(group.dataset(name)?.read_f64()?) -} - -/// Read the embeddings dataset as a flat `Vec` of `n * dim` values, -/// handling both f32 and (lossy) f16 storage. -fn read_embeddings_flat(group: &Group<'_>) -> Result, BoxErr> { - Ok(group.dataset("embeddings")?.read_f32()?) -} - -/// Read a migration HDF5 file into a [`SqliteData`]. -pub fn read_hdf5(path: &str) -> Result { - let file = File::open(path)?; - - let embedding_dim = match file.root().attrs()?.get("embedding_dim") { - Some(AttrValue::I64(d)) => *d as usize, - _ => 0, - }; - - let chunks = read_chunks(&file, embedding_dim)?; - let sessions = read_sessions(&file)?; - let entities = read_entities(&file)?; - let relations = read_relations(&file)?; - - Ok(SqliteData { - chunks, - sessions, - entities, - relations, - embedding_dim, - // Not a SQLite read — the caller (incremental migration) carries - // forward the current run's actual `source_path` from the fresh - // SQLite read instead of using this placeholder. - source_path: String::new(), - }) -} - -fn read_chunks(file: &File, dim: usize) -> Result, BoxErr> { - let g = file.group("chunks")?; - let count = group_count(&g)?; - if count == 0 { - return Ok(Vec::new()); - } - let ids = read_i64s(&g, "id")?; - let texts = read_strings(&g, "text")?; - let channels = read_strings(&g, "source_channel")?; - let timestamps = read_f64s(&g, "timestamp")?; - let session_ids = read_strings(&g, "session_id")?; - let tags = read_strings(&g, "tags")?; - let deleted = g.dataset("deleted")?.read_i32()?; - let emb_flat = read_embeddings_flat(&g)?; - let dim = dim.max(1); - - let mut chunks = Vec::with_capacity(ids.len()); - for (i, &id) in ids.iter().enumerate() { - let embedding = emb_flat - .get(i * dim..(i + 1) * dim) - .map(|s| s.to_vec()) - .unwrap_or_default(); - chunks.push(MemoryChunk { - id, - chunk: texts.get(i).cloned().unwrap_or_default(), - embedding, - source_channel: channels.get(i).cloned().unwrap_or_default(), - timestamp: timestamps.get(i).copied().unwrap_or(0.0), - session_id: session_ids.get(i).cloned().unwrap_or_default(), - tags: tags.get(i).cloned().unwrap_or_default(), - deleted: deleted.get(i).copied().unwrap_or(0), - }); - } - Ok(chunks) -} - -fn read_sessions(file: &File) -> Result, BoxErr> { - let g = file.group("sessions")?; - if group_count(&g)? == 0 { - return Ok(Vec::new()); - } - let ids = read_strings(&g, "id")?; - let starts = read_i64s(&g, "start_idx")?; - let ends = read_i64s(&g, "end_idx")?; - let channels = read_strings(&g, "channel")?; - let timestamps = read_f64s(&g, "timestamp")?; - let summaries = read_strings(&g, "summary")?; - Ok((0..ids.len()) - .map(|i| Session { - id: ids[i].clone(), - start_idx: starts.get(i).copied().unwrap_or(0), - end_idx: ends.get(i).copied().unwrap_or(0), - channel: channels.get(i).cloned().unwrap_or_default(), - timestamp: timestamps.get(i).copied().unwrap_or(0.0), - summary: summaries.get(i).cloned().unwrap_or_default(), - }) - .collect()) -} - -fn read_entities(file: &File) -> Result, BoxErr> { - let g = file.group("entities")?; - if group_count(&g)? == 0 { - return Ok(Vec::new()); - } - let ids = read_i64s(&g, "id")?; - let names = read_strings(&g, "name")?; - let types = read_strings(&g, "type")?; - let emb_idxs = read_i64s(&g, "embedding_idx")?; - Ok((0..ids.len()) - .map(|i| Entity { - id: ids[i], - name: names.get(i).cloned().unwrap_or_default(), - entity_type: types.get(i).cloned().unwrap_or_default(), - embedding_idx: emb_idxs.get(i).copied().unwrap_or(-1), - }) - .collect()) -} - -fn read_relations(file: &File) -> Result, BoxErr> { - let g = file.group("relations")?; - if group_count(&g)? == 0 { - return Ok(Vec::new()); - } - let srcs = read_i64s(&g, "src")?; - let tgts = read_i64s(&g, "tgt")?; - let rels = read_strings(&g, "relation")?; - let weights = read_f64s(&g, "weight")?; - let timestamps = read_f64s(&g, "timestamp")?; - Ok((0..srcs.len()) - .map(|i| Relation { - src: srcs[i], - tgt: tgts.get(i).copied().unwrap_or(0), - relation: rels.get(i).cloned().unwrap_or_default(), - weight: weights.get(i).copied().unwrap_or(1.0), - timestamp: timestamps.get(i).copied().unwrap_or(0.0), - }) - .collect()) -} - -fn group_count(group: &Group<'_>) -> Result { - match group.attrs()?.get("count") { - Some(AttrValue::I64(n)) => Ok(*n as u64), - _ => Ok(0), - } -} diff --git a/crates/clawhdf5-migrate/src/hdf5_writer.rs b/crates/clawhdf5-migrate/src/hdf5_writer.rs deleted file mode 100644 index 8cf1f18..0000000 --- a/crates/clawhdf5-migrate/src/hdf5_writer.rs +++ /dev/null @@ -1,366 +0,0 @@ -use clawhdf5::writer::FileBuilder; -use clawhdf5_format::datatype::{CharacterSet, Datatype, StringPadding}; -use clawhdf5_format::type_builders::AttrValue; - -use crate::sqlite_reader::SqliteData; - -/// Options controlling HDF5 output. -pub struct WriteOptions { - pub agent_id: String, - pub embedder: String, - pub compression: bool, - pub compression_level: u32, - pub float16: bool, -} - -/// Write SQLite data to an HDF5 file. -pub fn write_hdf5( - path: &str, - data: &SqliteData, - opts: &WriteOptions, -) -> Result<(), Box> { - let mut builder = FileBuilder::new(); - let timestamp = iso8601_now(); - - // Root-level metadata attributes - builder.set_attr("agent_id", AttrValue::String(opts.agent_id.clone())); - builder.set_attr("embedder", AttrValue::String(opts.embedder.clone())); - builder.set_attr("embedding_dim", AttrValue::I64(data.embedding_dim as i64)); - builder.set_attr("source", AttrValue::String("sqlite-migration".into())); - builder.set_attr("version", AttrValue::I64(1)); - // Lineage: which SQLite database this output was migrated from and when, - // plus the migrator tool version — so a chain of `--incremental` runs - // still has an audit trail instead of every run overwriting the same - // static attributes (see research/03_provenance.md, INT-03). - builder.set_attr("source_path", AttrValue::String(data.source_path.clone())); - builder.set_attr("migrated_at", AttrValue::String(timestamp.clone())); - builder.set_attr( - "migrator_version", - AttrValue::String(env!("CARGO_PKG_VERSION").to_owned()), - ); - - write_chunks_group(&mut builder, data, opts, ×tamp); - write_sessions_group(&mut builder, data); - write_entities_group(&mut builder, data); - write_relations_group(&mut builder, data); - - builder.write(path)?; - Ok(()) -} - -/// Current UTC time formatted as an ISO-8601 / RFC-3339 timestamp -/// (`YYYY-MM-DDTHH:MM:SSZ`), with no external date/time dependency. -fn iso8601_now() -> String { - let secs = std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .unwrap_or_default() - .as_secs(); - let days = (secs / 86_400) as i64; - let time_of_day = secs % 86_400; - let (h, m, s) = ( - time_of_day / 3600, - (time_of_day % 3600) / 60, - time_of_day % 60, - ); - let (y, mo, d) = civil_from_days(days); - format!("{y:04}-{mo:02}-{d:02}T{h:02}:{m:02}:{s:02}Z") -} - -/// Days-since-epoch to (year, month, day), Howard Hinnant's `civil_from_days` -/// algorithm (proleptic Gregorian calendar, valid for the full `i64` range). -fn civil_from_days(z: i64) -> (i64, u32, u32) { - let z = z + 719_468; - let era = if z >= 0 { z } else { z - 146_096 } / 146_097; - let doe = (z - era * 146_097) as u64; // [0, 146096] - let yoe = (doe - doe / 1460 + doe / 36_524 - doe / 146_096) / 365; // [0, 399] - let y = yoe as i64 + era * 400; - let doy = doe - (365 * yoe + yoe / 4 - yoe / 100); // [0, 365] - let mp = (5 * doy + 2) / 153; // [0, 11] - let d = (doy - (153 * mp + 2) / 5 + 1) as u32; // [1, 31] - let m = (if mp < 10 { mp + 3 } else { mp - 9 }) as u32; // [1, 12] - let y = if m <= 2 { y + 1 } else { y }; - (y, m, d) -} - -/// Build a fixed-length string Datatype from the max byte length of the items. -fn string_dtype(max_len: usize) -> Datatype { - Datatype::String { - size: max_len.max(1) as u32, - padding: StringPadding::NullPad, - charset: CharacterSet::Utf8, - } -} - -/// Pack a slice of strings into null-padded raw bytes of uniform width. -fn pack_strings(strings: &[String]) -> (Vec, usize) { - let max_len = strings.iter().map(|s| s.len()).max().unwrap_or(0).max(1); - let mut buf = vec![0u8; strings.len() * max_len]; - for (i, s) in strings.iter().enumerate() { - let start = i * max_len; - let bytes = s.as_bytes(); - let copy_len = bytes.len().min(max_len); - buf[start..start + copy_len].copy_from_slice(&bytes[..copy_len]); - } - (buf, max_len) -} - -fn apply_compression(ds: &mut clawhdf5_format::type_builders::DatasetBuilder, opts: &WriteOptions) { - if opts.compression { - ds.with_deflate(opts.compression_level); - ds.with_shuffle(); - } -} - -fn write_chunks_group( - builder: &mut FileBuilder, - data: &SqliteData, - opts: &WriteOptions, - timestamp: &str, -) { - let mut group = builder.create_group("chunks"); - let n = data.chunks.len() as u64; - - if n == 0 { - group.set_attr("count", AttrValue::I64(0)); - builder.add_group(group.finish()); - return; - } - - group.set_attr("count", AttrValue::I64(n as i64)); - - // Source attribution attached directly to the content-bearing datasets - // (SHA-256 of the raw bytes + creator/timestamp/source), so the chunk - // text and embeddings each carry their own verifiable provenance - // (see clawhdf5_format::provenance / `Dataset::verify_provenance`). - let source_opt = if data.source_path.is_empty() { - None - } else { - Some(data.source_path.as_str()) - }; - - // ids - let ids: Vec = data.chunks.iter().map(|c| c.id).collect(); - group.create_dataset("id").with_i64_data(&ids); - - // text - let texts: Vec = data.chunks.iter().map(|c| c.chunk.clone()).collect(); - let (text_raw, text_len) = pack_strings(&texts); - group - .create_dataset("text") - .with_compound_data(string_dtype(text_len), text_raw, n) - .with_provenance("clawhdf5-migrate", timestamp, source_opt); - - // embeddings - flatten to [N, dim] - let dim = data.embedding_dim; - if opts.float16 { - let f16_data: Vec = data - .chunks - .iter() - .flat_map(|c| { - c.embedding - .iter() - .map(|&v| half::f16::from_f32(v).to_bits()) - }) - .collect(); - let raw: Vec = f16_data.iter().flat_map(|v| v.to_le_bytes()).collect(); - let f16_dtype = Datatype::FloatingPoint { - size: 2, - byte_order: clawhdf5_format::datatype::DatatypeByteOrder::LittleEndian, - bit_offset: 0, - bit_precision: 16, - exponent_location: 10, - exponent_size: 5, - mantissa_location: 0, - mantissa_size: 10, - exponent_bias: 15, - }; - let ds = group - .create_dataset("embeddings") - .with_compound_data(f16_dtype, raw, n) - .with_shape(&[n, dim as u64]) - .with_provenance("clawhdf5-migrate", timestamp, source_opt); - apply_compression(ds, opts); - } else { - let flat: Vec = data - .chunks - .iter() - .flat_map(|c| c.embedding.iter().copied()) - .collect(); - let ds = group - .create_dataset("embeddings") - .with_f32_data(&flat) - .with_shape(&[n, dim as u64]) - .with_provenance("clawhdf5-migrate", timestamp, source_opt); - apply_compression(ds, opts); - } - - // source_channel - let channels: Vec = data - .chunks - .iter() - .map(|c| c.source_channel.clone()) - .collect(); - let (ch_raw, ch_len) = pack_strings(&channels); - group - .create_dataset("source_channel") - .with_compound_data(string_dtype(ch_len), ch_raw, n); - - // timestamp - let timestamps: Vec = data.chunks.iter().map(|c| c.timestamp).collect(); - group.create_dataset("timestamp").with_f64_data(×tamps); - - // session_id - let sess_ids: Vec = data.chunks.iter().map(|c| c.session_id.clone()).collect(); - let (sid_raw, sid_len) = pack_strings(&sess_ids); - group - .create_dataset("session_id") - .with_compound_data(string_dtype(sid_len), sid_raw, n); - - // tags - let tags: Vec = data.chunks.iter().map(|c| c.tags.clone()).collect(); - let (tag_raw, tag_len) = pack_strings(&tags); - group - .create_dataset("tags") - .with_compound_data(string_dtype(tag_len), tag_raw, n); - - // deleted - let deleted: Vec = data.chunks.iter().map(|c| c.deleted).collect(); - group.create_dataset("deleted").with_i32_data(&deleted); - - builder.add_group(group.finish()); -} - -fn write_sessions_group(builder: &mut FileBuilder, data: &SqliteData) { - let mut group = builder.create_group("sessions"); - let n = data.sessions.len() as u64; - group.set_attr("count", AttrValue::I64(n as i64)); - - if n == 0 { - builder.add_group(group.finish()); - return; - } - - let ids: Vec = data.sessions.iter().map(|s| s.id.clone()).collect(); - let (id_raw, id_len) = pack_strings(&ids); - group - .create_dataset("id") - .with_compound_data(string_dtype(id_len), id_raw, n); - - let start_idxs: Vec = data.sessions.iter().map(|s| s.start_idx).collect(); - group.create_dataset("start_idx").with_i64_data(&start_idxs); - - let end_idxs: Vec = data.sessions.iter().map(|s| s.end_idx).collect(); - group.create_dataset("end_idx").with_i64_data(&end_idxs); - - let channels: Vec = data.sessions.iter().map(|s| s.channel.clone()).collect(); - let (ch_raw, ch_len) = pack_strings(&channels); - group - .create_dataset("channel") - .with_compound_data(string_dtype(ch_len), ch_raw, n); - - let timestamps: Vec = data.sessions.iter().map(|s| s.timestamp).collect(); - group.create_dataset("timestamp").with_f64_data(×tamps); - - let summaries: Vec = data.sessions.iter().map(|s| s.summary.clone()).collect(); - let (sum_raw, sum_len) = pack_strings(&summaries); - group - .create_dataset("summary") - .with_compound_data(string_dtype(sum_len), sum_raw, n); - - builder.add_group(group.finish()); -} - -fn write_entities_group(builder: &mut FileBuilder, data: &SqliteData) { - let mut group = builder.create_group("entities"); - let n = data.entities.len() as u64; - group.set_attr("count", AttrValue::I64(n as i64)); - - if n == 0 { - builder.add_group(group.finish()); - return; - } - - let ids: Vec = data.entities.iter().map(|e| e.id).collect(); - group.create_dataset("id").with_i64_data(&ids); - - let names: Vec = data.entities.iter().map(|e| e.name.clone()).collect(); - let (name_raw, name_len) = pack_strings(&names); - group - .create_dataset("name") - .with_compound_data(string_dtype(name_len), name_raw, n); - - let types: Vec = data - .entities - .iter() - .map(|e| e.entity_type.clone()) - .collect(); - let (type_raw, type_len) = pack_strings(&types); - group - .create_dataset("type") - .with_compound_data(string_dtype(type_len), type_raw, n); - - let emb_idxs: Vec = data.entities.iter().map(|e| e.embedding_idx).collect(); - group - .create_dataset("embedding_idx") - .with_i64_data(&emb_idxs); - - builder.add_group(group.finish()); -} - -fn write_relations_group(builder: &mut FileBuilder, data: &SqliteData) { - let mut group = builder.create_group("relations"); - let n = data.relations.len() as u64; - group.set_attr("count", AttrValue::I64(n as i64)); - - if n == 0 { - builder.add_group(group.finish()); - return; - } - - let srcs: Vec = data.relations.iter().map(|r| r.src).collect(); - group.create_dataset("src").with_i64_data(&srcs); - - let tgts: Vec = data.relations.iter().map(|r| r.tgt).collect(); - group.create_dataset("tgt").with_i64_data(&tgts); - - let rels: Vec = data.relations.iter().map(|r| r.relation.clone()).collect(); - let (rel_raw, rel_len) = pack_strings(&rels); - group - .create_dataset("relation") - .with_compound_data(string_dtype(rel_len), rel_raw, n); - - let weights: Vec = data.relations.iter().map(|r| r.weight).collect(); - group.create_dataset("weight").with_f64_data(&weights); - - let timestamps: Vec = data.relations.iter().map(|r| r.timestamp).collect(); - group.create_dataset("timestamp").with_f64_data(×tamps); - - builder.add_group(group.finish()); -} - -#[cfg(test)] -mod time_tests { - use super::civil_from_days; - - #[test] - fn epoch_day_zero_is_1970_01_01() { - assert_eq!(civil_from_days(0), (1970, 1, 1)); - } - - #[test] - fn known_dates_roundtrip() { - // 2026-08-16 is 20,681 days after 1970-01-01. - assert_eq!(civil_from_days(20_681), (2026, 8, 16)); - // 2000-02-29 (leap day itself) and 2000-03-01 (the day after). - assert_eq!(civil_from_days(11_016), (2000, 2, 29)); - assert_eq!(civil_from_days(11_017), (2000, 3, 1)); - } - - #[test] - fn iso8601_now_has_expected_shape() { - let ts = super::iso8601_now(); - assert_eq!(ts.len(), "2026-08-16T00:00:00Z".len()); - assert!(ts.starts_with("20")); // sanity: 21st-century year - assert!(ts.ends_with('Z')); - } -} diff --git a/crates/clawhdf5-migrate/src/main.rs b/crates/clawhdf5-migrate/src/main.rs index 932f826..4a0f58b 100644 --- a/crates/clawhdf5-migrate/src/main.rs +++ b/crates/clawhdf5-migrate/src/main.rs @@ -1,13 +1,21 @@ -mod hdf5_reader; -mod hdf5_writer; 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; -/// Migrate ZeroClaw agent memory from SQLite to HDF5 format. +type BoxErr = Box; + +/// 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 ...`). #[derive(Parser, Debug)] #[command(name = "clawhdf5-migrate", version, about)] struct Cli { @@ -15,27 +23,32 @@ struct Cli { #[arg(long)] sqlite: String, - /// Destination HDF5 file path + /// Destination agent store (.h5). Replaced if it exists, unless + /// --incremental #[arg(long)] hdf5: String, - /// Agent ID for metadata + /// Agent ID recorded in the new store #[arg(long, default_value = "migrated")] agent_id: String, - /// Embedder name for metadata + /// Embedder name recorded in the new store #[arg(long, default_value = "unknown")] embedder: String, - /// Embedding dimension (auto-detect from first row if not specified) + /// 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, - /// Skip deleted/tombstoned entries + /// Skip deleted/tombstoned entries (otherwise they are migrated as + /// deleted records) #[arg(long)] skip_deleted: bool, - /// Enable deflate compression on embeddings + /// Compress the embeddings dataset (deflate) #[arg(long)] compression: bool, @@ -43,8 +56,14 @@ struct Cli { #[arg(long, default_value_t = 4)] compression_level: u32, - /// Store embeddings as float16 (halves storage) + /// 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 @@ -55,8 +74,12 @@ struct Cli { #[arg(long)] validate_full: bool, - /// Append only rows newer than the existing output (by chunk id), merging - /// into the file at --hdf5 if it exists + /// 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, @@ -99,9 +122,15 @@ fn schema_from_cli(cli: &Cli) -> SchemaConfig { c } -fn main() -> Result<(), Box> { - let cli = Cli::parse(); - let schema = schema_from_cli(&cli); +/// The result of a (non-dry) run. +#[derive(Debug)] +struct Outcome { + migration: Migration, + summary: ValidationSummary, +} + +fn run(cli: &Cli) -> Result, BoxErr> { + let schema = schema_from_cli(cli); // Dry run: a fast count-only pass that does not buffer the database. if cli.dry_run { @@ -111,111 +140,134 @@ fn main() -> Result<(), Box> { "Would migrate: {} chunks, {} sessions, {} entities, {} relations", counts.chunks, counts.sessions, counts.entities, counts.relations ); - return Ok(()); + 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); } - - // Incremental: merge new rows into the existing output (if present). - let incremental_base = if cli.incremental && std::path::Path::new(&cli.hdf5).exists() { - Some(hdf5_reader::read_hdf5(&cli.hdf5)?) - } else { - None - }; - let min_chunk_id = incremental_base - .as_ref() - .map(|d| d.chunks.iter().map(|c| c.id).max().unwrap_or(0)) - .unwrap_or(0); - let dim_hint = cli - .embedding_dim - .or_else(|| incremental_base.as_ref().map(|d| d.embedding_dim)); - - let source = if min_chunk_id > 0 { - sqlite_reader::read_sqlite_filtered( - &cli.sqlite, - cli.skip_deleted, - dim_hint, - &schema, - min_chunk_id, - )? - } else { - sqlite_reader::read_sqlite(&cli.sqlite, cli.skip_deleted, dim_hint, &schema)? - }; - - // Build the dataset to write: either the source alone, or the existing - // output plus the newly-read rows (metadata groups refreshed from source). - let data = match incremental_base { - Some(mut base) => { - let added = source.chunks.len(); - base.chunks.extend(source.chunks); - base.sessions = source.sessions; - base.entities = source.entities; - base.relations = source.relations; - base.embedding_dim = source.embedding_dim.max(base.embedding_dim); - // Carry the current run's real SQLite source forward for - // provenance — `base` (re-read from the prior HDF5 output) has - // no meaningful source_path of its own. - base.source_path = source.source_path; - if cli.verbose { - eprintln!("Incremental: appended {added} new chunks (id > {min_chunk_id})"); - } - base - } - None => source, - }; - + let data = + sqlite_reader::read_sqlite(&cli.sqlite, reader_skip_deleted, cli.embedding_dim, &schema)?; if cli.verbose { eprintln!( - "Migrating {} chunks, {} sessions, {} entities, {} relations (dim={})", + "Read {} chunks, {} sessions, {} entities, {} relations (dim={})", data.chunks.len(), data.sessions.len(), data.entities.len(), data.relations.len(), data.embedding_dim ); - eprintln!("Writing HDF5 file: {}", cli.hdf5); + eprintln!("Writing agent store: {}", cli.hdf5); } - let opts = hdf5_writer::WriteOptions { - agent_id: cli.agent_id, - embedder: cli.embedder, + 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), - float16: cli.float16, + f32: cli.f32, + incremental: cli.incremental, + skip_deleted: cli.skip_deleted, }; + let migration = store_writer::write_store(out, &data, &opts)?; - hdf5_writer::write_hdf5(&cli.hdf5, &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 { - eprintln!("Validating output (content check)..."); + 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_hdf5(&cli.hdf5, &data, cli.validate_full, cli.float16)?; + 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: {} chunks, {} sessions, {} entities, {} relations (dim={}); {} rows content-verified", - summary.chunks, + "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 { + "" + }, ); - if summary.provenance_verified { - eprintln!("Provenance: chunks/text and chunks/embeddings SHA-256 hashes verified."); - } else if cli.verbose { - eprintln!("Provenance: no provenance hash found to verify (older output format?)."); - } - 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. @@ -303,12 +355,57 @@ mod tests { (0..dim).map(|i| seed + i as f32 * 0.1).collect() } - // ---------- Test 1: Basic end-to-end migration ---------- + /// A unit-length embedding pointing mostly along axis `axis`, so records + /// are distinguishable by vector search. + fn axis_embedding(dim: usize, axis: usize) -> Vec { + let mut v: Vec = (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::().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 --hdf5 `. + fn migrate(db: &str, out: &Path, extra: &[&str]) -> Result, 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 = dir.path().join("out.h5"); + 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); @@ -318,32 +415,193 @@ mod tests { insert_relation(&conn, 1, 1, "self"); drop(conn); - let data = - sqlite_reader::read_sqlite(&db_path, false, None, &SchemaConfig::default()).unwrap(); - let opts = hdf5_writer::WriteOptions { - agent_id: "test-agent".into(), - embedder: "test-embed".into(), - compression: false, - compression_level: 4, - float16: false, - }; - hdf5_writer::write_hdf5(h5_path.to_str().unwrap(), &data, &opts).unwrap(); - - let summary = - validate::validate_hdf5(h5_path.to_str().unwrap(), &data, false, false).unwrap(); - assert_eq!(summary.chunks, 2); - assert_eq!(summary.sessions, 1); - assert_eq!(summary.entities, 1); - assert_eq!(summary.relations, 1); - assert_eq!(summary.embedding_dim, 8); + 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); } - // ---------- Test 2: Skip deleted rows ---------- + // ---------- 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 = c + .embedding + .iter() + .map(|&v| if expect_float16 { round_to_f16(v) } else { v }.to_bits()) + .collect(); + let got: Vec = 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 = dir.path().join("out.h5"); + 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); @@ -351,41 +609,31 @@ mod tests { insert_chunk(&conn, 3, "also active", &make_embedding(4, 3.0), 0); drop(conn); - let data = - sqlite_reader::read_sqlite(&db_path, true, None, &SchemaConfig::default()).unwrap(); - assert_eq!(data.chunks.len(), 2); - - let opts = hdf5_writer::WriteOptions { - agent_id: "t".into(), - embedder: "t".into(), - compression: false, - compression_level: 4, - float16: false, - }; - hdf5_writer::write_hdf5(h5_path.to_str().unwrap(), &data, &opts).unwrap(); - - let summary = - validate::validate_hdf5(h5_path.to_str().unwrap(), &data, false, false).unwrap(); - assert_eq!(summary.chunks, 2); + let o = migrate_ok(&db_path, &h5_path, &["--skip-deleted"]); + assert_eq!(o.summary.count, 2); + assert_eq!(o.summary.active, 2); } - // ---------- Test 3: Include deleted rows ---------- + // ---------- 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 data = - sqlite_reader::read_sqlite(&db_path, false, None, &SchemaConfig::default()).unwrap(); - assert_eq!(data.chunks.len(), 2); + 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]); } - // ---------- Test 4: Auto-detect embedding dimension ---------- + // ---------- Auto-detect embedding dimension ---------- #[test] fn test_auto_detect_dim() { let dir = TempDir::new().unwrap(); @@ -400,11 +648,12 @@ mod tests { assert_eq!(data.embedding_dim, 16); } - // ---------- Test 5: Manual embedding dimension ---------- + // ---------- 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); @@ -413,56 +662,225 @@ mod tests { let data = sqlite_reader::read_sqlite(&db_path, false, Some(8), &SchemaConfig::default()).unwrap(); assert_eq!(data.embedding_dim, 8); - // Embedding truncated to dim 8 - assert_eq!(data.chunks[0].embedding.len(), 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}"); } - // ---------- Test 6: Float16 conversion ---------- + // ---------- 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 = dir.path().join("out.h5"); + let h5_path = out_path(&dir, "out.h5"); let conn = Connection::open(&db_path).unwrap(); - let emb = vec![1.0f32, 2.5, -0.5, 3.125]; + // 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 data = - sqlite_reader::read_sqlite(&db_path, false, None, &SchemaConfig::default()).unwrap(); - let opts = hdf5_writer::WriteOptions { - agent_id: "t".into(), - embedder: "t".into(), - compression: false, - compression_level: 4, - float16: true, - }; - hdf5_writer::write_hdf5(h5_path.to_str().unwrap(), &data, &opts).unwrap(); - - // Content-validate with the float16 tolerance enabled. - let summary = - validate::validate_hdf5(h5_path.to_str().unwrap(), &data, true, true).unwrap(); - assert_eq!(summary.chunks, 1); - - // Verify float16 values are within tolerance - for &v in &emb { - let f16 = half::f16::from_f32(v); - let roundtrip = f16.to_f32(); - assert!( - (v - roundtrip).abs() < 0.01, - "f16 roundtrip too lossy for {v}" - ); + 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"); } - // ---------- Test 7: Compression produces valid file ---------- + // ---------- 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 = dir.path().join("compressed.h5"); - let h5_uncompressed = dir.path().join("uncompressed.h5"); + 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 @@ -471,27 +889,17 @@ mod tests { } drop(conn); - let data = - sqlite_reader::read_sqlite(&db_path, false, None, &SchemaConfig::default()).unwrap(); - - let opts_compressed = hdf5_writer::WriteOptions { - agent_id: "t".into(), - embedder: "t".into(), - compression: true, - compression_level: 6, - float16: false, - }; - hdf5_writer::write_hdf5(h5_compressed.to_str().unwrap(), &data, &opts_compressed).unwrap(); - - let opts_plain = hdf5_writer::WriteOptions { - agent_id: "t".into(), - embedder: "t".into(), - compression: false, - compression_level: 4, - float16: false, - }; - hdf5_writer::write_hdf5(h5_uncompressed.to_str().unwrap(), &data, &opts_plain).unwrap(); + 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!( @@ -500,30 +908,31 @@ mod tests { ); } - // ---------- Test 8: Dry run doesn't create file ---------- + // ---------- 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 = dir.path().join("should_not_exist.h5"); + 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); - // Simulate dry-run: read data but don't write - let data = - sqlite_reader::read_sqlite(&db_path, false, None, &SchemaConfig::default()).unwrap(); - assert_eq!(data.chunks.len(), 1); + assert!( + migrate(&db_path, &h5_path, &["--dry-run"]) + .unwrap() + .is_none() + ); assert!(!h5_path.exists()); } - // ---------- Test 9: Empty database migration ---------- + // ---------- Empty database migration ---------- #[test] fn test_empty_db() { let dir = TempDir::new().unwrap(); let db_path = create_test_db(&dir); - let h5_path = dir.path().join("out.h5"); + let h5_path = out_path(&dir, "out.h5"); let data = sqlite_reader::read_sqlite(&db_path, false, None, &SchemaConfig::default()).unwrap(); @@ -532,26 +941,51 @@ mod tests { assert_eq!(data.entities.len(), 0); assert_eq!(data.relations.len(), 0); - let opts = hdf5_writer::WriteOptions { - agent_id: "t".into(), - embedder: "t".into(), - compression: false, - compression_level: 4, - float16: false, - }; - hdf5_writer::write_hdf5(h5_path.to_str().unwrap(), &data, &opts).unwrap(); + // 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 summary = - validate::validate_hdf5(h5_path.to_str().unwrap(), &data, false, false).unwrap(); - assert_eq!(summary.chunks, 0); + 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); } - // ---------- Test 10: Large migration (1000 entries) ---------- + // ---------- 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 = dir.path().join("out.h5"); + let h5_path = out_path(&dir, "out.h5"); let conn = Connection::open(&db_path).unwrap(); for i in 0..1000 { @@ -565,30 +999,21 @@ mod tests { } drop(conn); - let data = - sqlite_reader::read_sqlite(&db_path, false, None, &SchemaConfig::default()).unwrap(); - assert_eq!(data.chunks.len(), 1000); - - let opts = hdf5_writer::WriteOptions { - agent_id: "t".into(), - embedder: "t".into(), - compression: false, - compression_level: 4, - float16: false, - }; - hdf5_writer::write_hdf5(h5_path.to_str().unwrap(), &data, &opts).unwrap(); - - let summary = - validate::validate_hdf5(h5_path.to_str().unwrap(), &data, false, false).unwrap(); - assert_eq!(summary.chunks, 1000); + 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); } - // ---------- Test 11: Session migration ---------- + // ---------- Session migration ---------- #[test] fn test_session_migration() { let dir = TempDir::new().unwrap(); let db_path = create_test_db(&dir); - let h5_path = dir.path().join("out.h5"); + let h5_path = out_path(&dir, "out.h5"); let conn = Connection::open(&db_path).unwrap(); insert_session(&conn, "session-alpha", 0, 10); @@ -596,30 +1021,28 @@ mod tests { insert_session(&conn, "session-gamma", 21, 30); drop(conn); - let data = - sqlite_reader::read_sqlite(&db_path, false, None, &SchemaConfig::default()).unwrap(); - assert_eq!(data.sessions.len(), 3); - - let opts = hdf5_writer::WriteOptions { - agent_id: "t".into(), - embedder: "t".into(), - compression: false, - compression_level: 4, - float16: false, - }; - hdf5_writer::write_hdf5(h5_path.to_str().unwrap(), &data, &opts).unwrap(); - - let summary = - validate::validate_hdf5(h5_path.to_str().unwrap(), &data, false, false).unwrap(); - assert_eq!(summary.sessions, 3); + 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") + ); } - // ---------- Test 12: Knowledge graph (entities + relations) ---------- + // ---------- 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 = dir.path().join("out.h5"); + let h5_path = out_path(&dir, "out.h5"); let conn = Connection::open(&db_path).unwrap(); insert_entity(&conn, 1, "Alice", "person"); @@ -630,145 +1053,102 @@ mod tests { insert_relation(&conn, 2, 3, "uses"); drop(conn); - let data = - sqlite_reader::read_sqlite(&db_path, false, None, &SchemaConfig::default()).unwrap(); - assert_eq!(data.entities.len(), 3); - assert_eq!(data.relations.len(), 3); - - let opts = hdf5_writer::WriteOptions { - agent_id: "t".into(), - embedder: "t".into(), - compression: false, - compression_level: 4, - float16: false, - }; - hdf5_writer::write_hdf5(h5_path.to_str().unwrap(), &data, &opts).unwrap(); - - let summary = - validate::validate_hdf5(h5_path.to_str().unwrap(), &data, false, false).unwrap(); - assert_eq!(summary.entities, 3); - assert_eq!(summary.relations, 3); + 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"]); } - // ---------- Test 13: Validation catches chunk count mismatch ---------- + // ---------- 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 = dir.path().join("out.h5"); + 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 = - sqlite_reader::read_sqlite(&db_path, false, None, &SchemaConfig::default()).unwrap(); - let opts = hdf5_writer::WriteOptions { - agent_id: "t".into(), - embedder: "t".into(), - compression: false, - compression_level: 4, - float16: false, - }; - hdf5_writer::write_hdf5(h5_path.to_str().unwrap(), &data, &opts).unwrap(); - - // Validating against a source with an extra (unwritten) chunk must fail. - let mut bigger = - sqlite_reader::read_sqlite(&db_path, false, None, &SchemaConfig::default()).unwrap(); - let mut extra = bigger.chunks[0].clone(); - extra.id = 999; - bigger.chunks.push(extra); - let result = validate::validate_hdf5(h5_path.to_str().unwrap(), &bigger, false, false); - assert!(result.is_err()); + 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")); } - // ---------- Test 14: Metadata attributes are stored ---------- + // ---------- 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 = dir.path().join("out.h5"); + 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); - let data = - sqlite_reader::read_sqlite(&db_path, false, None, &SchemaConfig::default()).unwrap(); - let opts = hdf5_writer::WriteOptions { - agent_id: "my-agent-42".into(), - embedder: "openai-ada".into(), - compression: false, - compression_level: 4, - float16: false, - }; - hdf5_writer::write_hdf5(h5_path.to_str().unwrap(), &data, &opts).unwrap(); - - let file = clawhdf5::File::open(h5_path.to_str().unwrap()).unwrap(); - let root = file.root(); - let attrs = root.attrs().unwrap(); - - match attrs.get("agent_id") { - Some(clawhdf5_format::type_builders::AttrValue::String(s)) => { - assert_eq!(s, "my-agent-42"); - } - other => panic!("Expected String agent_id, got {other:?}"), - } - match attrs.get("embedder") { - Some(clawhdf5_format::type_builders::AttrValue::String(s)) => { - assert_eq!(s, "openai-ada"); - } - other => panic!("Expected String embedder, got {other:?}"), - } - match attrs.get("embedding_dim") { - Some(clawhdf5_format::type_builders::AttrValue::I64(d)) => { - assert_eq!(*d, 8); - } - other => panic!("Expected I64 embedding_dim, got {other:?}"), - } + 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); } - // ---------- Test 15: Embedding values roundtrip correctly ---------- + // ---------- 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 = dir.path().join("out.h5"); + 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); - let data = - sqlite_reader::read_sqlite(&db_path, false, None, &SchemaConfig::default()).unwrap(); - let opts = hdf5_writer::WriteOptions { - agent_id: "t".into(), - embedder: "t".into(), - compression: false, - compression_level: 4, - float16: false, - }; - hdf5_writer::write_hdf5(h5_path.to_str().unwrap(), &data, &opts).unwrap(); - - let file = clawhdf5::File::open(h5_path.to_str().unwrap()).unwrap(); - let chunks_group = file.group("chunks").unwrap(); - let emb_ds = chunks_group.dataset("embeddings").unwrap(); - let read_back = emb_ds.read_f32().unwrap(); - - assert_eq!(read_back.len(), 4); - for (a, b) in emb.iter().zip(read_back.iter()) { - assert!((a - b).abs() < 1e-6, "Embedding mismatch: {a} vs {b}"); - } + migrate_ok(&db_path, &h5_path, &["--f32"]); + let mem = HDF5Memory::open_read_only(&h5_path).unwrap(); + assert_eq!(&mem.cache.embeddings[0], &emb[..]); } - // ---------- Test 16: Full combined migration ---------- + // ---------- Full combined migration ---------- #[test] fn test_full_combined_migration() { let dir = TempDir::new().unwrap(); let db_path = create_test_db(&dir); - let h5_path = dir.path().join("out.h5"); + let h5_path = out_path(&dir, "out.h5"); let conn = Connection::open(&db_path).unwrap(); for i in 0..5 { @@ -787,89 +1167,74 @@ mod tests { insert_relation(&conn, 1, 2, "knows"); drop(conn); - // Skip deleted - let data = - sqlite_reader::read_sqlite(&db_path, true, None, &SchemaConfig::default()).unwrap(); - assert_eq!(data.chunks.len(), 4); // chunk 3 is deleted - - let opts = hdf5_writer::WriteOptions { - agent_id: "combined-test".into(), - embedder: "test-embedder".into(), - compression: true, - compression_level: 4, - float16: false, - }; - hdf5_writer::write_hdf5(h5_path.to_str().unwrap(), &data, &opts).unwrap(); - - let summary = - validate::validate_hdf5(h5_path.to_str().unwrap(), &data, false, false).unwrap(); - assert_eq!(summary.chunks, 4); - assert_eq!(summary.sessions, 2); - assert_eq!(summary.entities, 2); - assert_eq!(summary.relations, 1); - assert_eq!(summary.embedding_dim, 16); + 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); } - // ---------- Test 17: Validation catches session mismatch ---------- + // ---------- 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 = dir.path().join("out.h5"); + 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 data = - sqlite_reader::read_sqlite(&db_path, false, None, &SchemaConfig::default()).unwrap(); - let opts = hdf5_writer::WriteOptions { - agent_id: "t".into(), - embedder: "t".into(), - compression: false, - compression_level: 4, - float16: false, - }; - hdf5_writer::write_hdf5(h5_path.to_str().unwrap(), &data, &opts).unwrap(); - - // Validating against a source whose session content differs must fail. - let mut tampered = - sqlite_reader::read_sqlite(&db_path, false, None, &SchemaConfig::default()).unwrap(); - tampered.sessions[0].summary = "DIFFERENT".into(); - let result = validate::validate_hdf5(h5_path.to_str().unwrap(), &tampered, false, false); - assert!(result.is_err()); + 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")); } - // ---------- Real content validation catches corrupt embeddings ---------- + // ---------- 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 = dir.path().join("out.h5"); + 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); - let data = - sqlite_reader::read_sqlite(&db_path, false, None, &SchemaConfig::default()).unwrap(); - let opts = hdf5_writer::WriteOptions { - agent_id: "t".into(), - embedder: "t".into(), - compression: false, - compression_level: 4, - float16: false, - }; - hdf5_writer::write_hdf5(h5_path.to_str().unwrap(), &data, &opts).unwrap(); - - // A source whose embedding differs (but counts match) must fail validation. - let mut tampered = - sqlite_reader::read_sqlite(&db_path, false, None, &SchemaConfig::default()).unwrap(); - tampered.chunks[0].embedding[3] += 9.0; - let result = validate::validate_hdf5(h5_path.to_str().unwrap(), &tampered, true, false); - assert!(result.is_err()); - assert!(result.unwrap_err().to_string().contains("embedding")); + 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 ---------- @@ -911,6 +1276,15 @@ mod tests { // 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 ---------- @@ -918,44 +1292,290 @@ mod tests { fn test_incremental_migration() { let dir = TempDir::new().unwrap(); let db_path = create_test_db(&dir); - let h5_path = dir.path().join("out.h5"); - let cfg = SchemaConfig::default(); - let opts = hdf5_writer::WriteOptions { - agent_id: "t".into(), - embedder: "t".into(), - compression: false, - compression_level: 4, - float16: false, - }; + let h5_path = out_path(&dir, "out.h5"); - // First migration: 2 chunks. + // 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 data = sqlite_reader::read_sqlite(&db_path, false, None, &cfg).unwrap(); - hdf5_writer::write_hdf5(h5_path.to_str().unwrap(), &data, &opts).unwrap(); + let o = migrate_ok(&db_path, &h5_path, &["--incremental", "--f32"]); + assert!(!o.migration.appended_to_existing); + assert_eq!(o.summary.count, 2); - // Add two more rows, then migrate incrementally. + // 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), 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); - let base = hdf5_reader::read_hdf5(h5_path.to_str().unwrap()).unwrap(); - let max_id = base.chunks.iter().map(|c| c.id).max().unwrap_or(0); - assert_eq!(max_id, 2); - let new = - sqlite_reader::read_sqlite_filtered(&db_path, false, Some(4), &cfg, max_id).unwrap(); - assert_eq!(new.chunks.len(), 2); // only id 3 and 4 + // --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 mut merged = base; - merged.chunks.extend(new.chunks); - hdf5_writer::write_hdf5(h5_path.to_str().unwrap(), &merged, &opts).unwrap(); + 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); - let final_data = hdf5_reader::read_hdf5(h5_path.to_str().unwrap()).unwrap(); - assert_eq!(final_data.chunks.len(), 4); - let texts: Vec<&str> = final_data.chunks.iter().map(|c| c.chunk.as_str()).collect(); - assert_eq!(texts, vec!["one", "two", "three", "four"]); + // 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::>() + .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) + ); + } } } diff --git a/crates/clawhdf5-migrate/src/sqlite_reader.rs b/crates/clawhdf5-migrate/src/sqlite_reader.rs index 26c04e6..1a20dbe 100644 --- a/crates/clawhdf5-migrate/src/sqlite_reader.rs +++ b/crates/clawhdf5-migrate/src/sqlite_reader.rs @@ -50,12 +50,8 @@ pub struct SqliteData { pub sessions: Vec, pub entities: Vec, pub relations: Vec, + /// `--embedding-dim`, or the first row's; 0 when neither exists. pub embedding_dim: usize, - /// Filesystem path of the SQLite database this data was read from, for - /// provenance attribution on the HDF5 output. Empty when the data did - /// not come directly from a SQLite read (e.g. re-read of a prior HDF5 - /// migration output for an incremental merge). - pub source_path: String, } /// A table name plus the ordered column names the reader maps by position. @@ -167,11 +163,13 @@ pub fn read_counts( }) } -/// Auto-detect embedding dimension from the first chunk's BLOB size. +/// Auto-detect embedding dimension from the BLOB size of the first chunk (in +/// id order, deleted or not). fn detect_embedding_dim(conn: &Connection, config: &SchemaConfig) -> SqlResult> { let emb_col = config.chunks.columns.get(2).copied().unwrap_or("embedding"); + let id_col = config.chunks.columns.first().copied().unwrap_or("id"); let mut stmt = conn.prepare(&format!( - "SELECT {emb_col} FROM {} LIMIT 1", + "SELECT {emb_col} FROM {} ORDER BY {id_col} LIMIT 1", config.chunks.table ))?; let mut rows = stmt.query([])?; @@ -195,24 +193,16 @@ fn blob_to_f32(blob: &[u8]) -> Vec { /// Read all data from a ZeroClaw SQLite database. /// /// If `skip_deleted` is true, rows with `deleted=1` are excluded from chunks. -/// If `embedding_dim` is `None`, auto-detect from the first row. +/// If `embedding_dim` is `None`, auto-detect from the first row (0 when there +/// are no rows). Embeddings are returned at their full stored length whatever +/// the dimension: checking that every row matches it is the writer's job +/// (`store_writer::write_store`), so a mismatch is an error, not silent +/// truncation. pub fn read_sqlite( path: &str, skip_deleted: bool, embedding_dim: Option, config: &SchemaConfig, -) -> Result> { - read_sqlite_filtered(path, skip_deleted, embedding_dim, config, 0) -} - -/// Like [`read_sqlite`] but only reads chunks whose id is greater than -/// `min_chunk_id` (0 = all). Used for incremental migration. -pub fn read_sqlite_filtered( - path: &str, - skip_deleted: bool, - embedding_dim: Option, - config: &SchemaConfig, - min_chunk_id: i64, ) -> Result> { let conn = Connection::open(path)?; @@ -221,7 +211,7 @@ pub fn read_sqlite_filtered( None => detect_embedding_dim(&conn, config)?.unwrap_or(0), }; - let chunks = read_chunks(&conn, skip_deleted, dim, config, min_chunk_id)?; + let chunks = read_chunks(&conn, skip_deleted, config)?; let sessions = read_sessions(&conn, config)?; let entities = read_entities(&conn, config)?; let relations = read_relations(&conn, config)?; @@ -232,42 +222,43 @@ pub fn read_sqlite_filtered( entities, relations, embedding_dim: dim, - source_path: path.to_owned(), }) } fn read_chunks( conn: &Connection, skip_deleted: bool, - expected_dim: usize, config: &SchemaConfig, - min_chunk_id: i64, ) -> SqlResult> { let id_col = config.chunks.columns.first().copied().unwrap_or("id"); let deleted_col = config.chunks.columns.get(7).copied().unwrap_or("deleted"); - let mut conds = Vec::new(); + let mut where_clause = String::new(); if skip_deleted { - conds.push(format!("{deleted_col} = 0")); + where_clause = format!(" WHERE {deleted_col} = 0"); } - if min_chunk_id > 0 { - conds.push(format!("{id_col} > {min_chunk_id}")); - } - let where_clause = if conds.is_empty() { - String::new() - } else { - format!(" WHERE {}", conds.join(" AND ")) - }; + // In id order, so the store's records follow the source's order. + where_clause.push_str(&format!(" ORDER BY {id_col}")); let sql = config.chunks.select(&where_clause); let mut stmt = conn.prepare(&sql)?; let rows = stmt.query_map([], |row| { let blob: Vec = row.get(2)?; - let mut embedding = blob_to_f32(&blob); - - // Validate/truncate to expected dimension - if expected_dim > 0 { - embedding.truncate(expected_dim); + if !blob.len().is_multiple_of(4) { + let id: i64 = row.get(0)?; + return Err(rusqlite::Error::FromSqlConversionFailure( + 2, + rusqlite::types::Type::Blob, + format!( + "chunk id {id}: embedding BLOB is {} bytes, not a whole number of \ + little-endian f32 values", + blob.len() + ) + .into(), + )); } + // Read at full length: rows of the wrong dimension are rejected by + // the writer, never truncated to fit. + let embedding = blob_to_f32(&blob); Ok(MemoryChunk { id: row.get(0)?, diff --git a/crates/clawhdf5-migrate/src/store_writer.rs b/crates/clawhdf5-migrate/src/store_writer.rs new file mode 100644 index 0000000..a707596 --- /dev/null +++ b/crates/clawhdf5-migrate/src/store_writer.rs @@ -0,0 +1,407 @@ +//! Write migrated SQLite data into a clawhdf5-agent store. +//! +//! Everything goes through `clawhdf5-agent`'s own API — `HDF5Memory::create` +//! (or `open` for `--incremental`), `save_batch`, `delete_batch`, the session +//! cache and the knowledge graph — so the result is an ordinary agent store +//! that `HDF5Memory::open` accepts, not a second hand-built copy of its schema. + +use std::collections::{HashMap, HashSet}; +use std::path::Path; + +use clawhdf5_agent::{AgentMemory, HDF5Memory, MemoryConfig, MemoryEntry}; +use clawhdf5_format::float16::round_to_f16; + +use crate::sqlite_reader::{MemoryChunk, SqliteData}; + +type BoxErr = Box; + +/// SQLite timestamps are Unix seconds; the agent's session and relation +/// timestamps are Unix microseconds (memory records stay in seconds). +pub const US_PER_SEC: f64 = 1_000_000.0; + +/// Options controlling the output store. +#[derive(Debug, Clone)] +pub struct WriteOptions { + pub agent_id: String, + pub embedder: String, + pub compression: bool, + pub compression_level: u32, + /// Store full-precision `f32` embeddings instead of the library default + /// (half precision). Only applies to a newly created store: an existing + /// store keeps the precision it was created with. + pub f32: bool, + /// Add to the store at the output path if there is one, instead of + /// replacing it. + pub incremental: bool, + /// Leave out deleted source rows that are not in the store. (A deleted + /// row that matches an active store record still tombstones it, so pass + /// deleted rows in `data` for an incremental run.) + pub skip_deleted: bool, +} + +/// What the migration wrote, and where each source row went, so validation +/// can compare the store with the source row by row. +#[derive(Debug, Default)] +pub struct Migration { + /// Whether the output store existed and was added to (`--incremental`). + pub appended_to_existing: bool, + /// The store's embedding precision. + pub float16: bool, + pub embedding_dim: usize, + /// Records in the store after the migration (including tombstones). + pub store_count: usize, + /// `(store index, source chunk index)` of every record written. + pub records: Vec<(usize, usize)>, + /// Source chunks already in the store (incremental), not written again. + pub chunks_present: usize, + /// `(store index, source chunk index)` of records that were active in + /// the store but whose source row is now deleted (incremental): they were + /// tombstoned by this run. + pub deleted_in_store: Vec<(usize, usize)>, + /// Source rows that were deleted in the store but are active in the + /// source (incremental): the agent has no un-delete, so each was written + /// again as a new record (counted in `records` too). + pub restored: usize, + /// Deleted source rows left out because of `skip_deleted`. + pub deleted_skipped: usize, + /// `(store session index, source session index)` of each session written. + pub sessions: Vec<(usize, usize)>, + pub sessions_present: usize, + /// `(store entity id, source entity index)` of each entity written. + pub entities: Vec<(u64, usize)>, + pub entities_present: usize, + /// SQLite entity id -> store entity id, for every source entity. + pub entity_ids: HashMap, + /// `(store relation index, source relation index)` of each relation written. + pub relations: Vec<(usize, usize)>, + pub relations_present: usize, + /// Source relations naming an entity id that is not in the entities + /// table; the knowledge graph cannot hold them, so they are skipped. + pub dangling_relations: Vec, + /// Messages of the write-anomaly alerts the agent raised while importing + /// (informational; they never block a save — a bulk import typically + /// trips the write-rate check). + pub anomaly_alerts: Vec, +} + +/// Identity of a memory record for incremental de-duplication: every field +/// the agent stores except the embedding (whose stored form depends on the +/// store's precision). +type RecordKey = (String, String, String, String, u64); + +fn record_key( + chunk: &str, + source_channel: &str, + session_id: &str, + tags: &str, + ts: f64, +) -> RecordKey { + ( + chunk.to_owned(), + source_channel.to_owned(), + session_id.to_owned(), + tags.to_owned(), + ts.to_bits(), + ) +} + +/// Reject rows the agent would otherwise store differently from the source, +/// or not at all: an embedding of a different length from the store's +/// dimension (the agent pads/truncates silently), an empty embedding, or, in +/// a float16 store, a value beyond the half-precision range. +/// +/// Every source row is checked, including ones that end up not being written +/// (already in the store, or deleted and skipped): the source must be +/// consistent as a whole, and the check runs before the store is touched. +fn check_chunks(chunks: &[MemoryChunk], dim: usize, float16: bool) -> Result<(), BoxErr> { + for c in chunks { + if c.embedding.is_empty() { + return Err(format!( + "chunk id {}: the embedding is empty; an agent store needs an embedding \ + for every record", + c.id + ) + .into()); + } + if c.embedding.len() != dim { + return Err(format!( + "chunk id {}: embedding has {} values, expected {dim}; every row must have \ + the store's dimension (detected from the first row unless --embedding-dim \ + is given), and rows are never truncated or padded to fit", + c.id, + c.embedding.len() + ) + .into()); + } + if float16 + && let Some((k, v)) = c + .embedding + .iter() + .enumerate() + .find(|&(_, &v)| v.is_finite() && round_to_f16(v).is_infinite()) + { + return Err(format!( + "chunk id {}: embedding[{k}] = {v} is outside the half-precision range \ + (±65504) of a float16 store; migrate with --f32", + c.id + ) + .into()); + } + } + Ok(()) +} + +/// Migrate `data` into the agent store at `path`. +/// +/// Without `opts.incremental` (or when nothing exists at `path`) a new store +/// is created, replacing any file there — but only once every source row has +/// passed [`check_chunks`], so a source that cannot be migrated leaves an +/// existing store untouched. With it, the existing store is opened and only +/// source rows it does not already hold are added: memory records are +/// matched on their content, sessions on their id, entities on name and +/// type, relations on (source, target, relation). A matched record then +/// takes the source row's deleted flag: see [`Migration::deleted_in_store`] +/// and [`Migration::restored`]. +pub fn write_store( + path: &Path, + data: &SqliteData, + opts: &WriteOptions, +) -> Result { + let existing = opts.incremental && path.exists(); + let mut mem = if existing { + // `open` does not modify the store beyond what the agent itself does + // on open; the checks below run before anything is written. + let mem = HDF5Memory::open(path)?; + let dim = mem.config().embedding_dim; + // `data.embedding_dim` is 0 only for a source with no records and no + // --embedding-dim, which has no dimension to disagree with. + if data.embedding_dim != 0 && dim != data.embedding_dim { + let hint = if dim == 0 { + " (a store created from a source with no memory records; re-create it \ + with --embedding-dim)" + } else { + "" + }; + return Err(format!( + "the store at {} has embedding_dim {dim}{hint}, the source {}; \ + embeddings of a different dimension cannot be added to it", + path.display(), + data.embedding_dim + ) + .into()); + } + check_chunks(&data.chunks, dim, mem.config().float16)?; + mem + } else { + // (With records, a dimension of 0 means an empty first embedding, + // which `check_chunks` reports more precisely.) + if data.embedding_dim == 0 && data.chunks.is_empty() { + return Err( + "the source has no memory records to detect the embedding dimension \ + from; pass --embedding-dim (the dimension of the agent's embedder), \ + or the store could never hold a record" + .into(), + ); + } + let mut config = MemoryConfig::new(path.to_path_buf(), &opts.agent_id, data.embedding_dim); + config.embedder = opts.embedder.clone(); + config.compression = opts.compression; + config.compression_level = opts.compression_level; + // Only ever switch the library default off (as `clawhdf5-cli create`). + if opts.f32 { + config.float16 = false; + } + // Before `create`, which replaces whatever is at `path`. + check_chunks(&data.chunks, config.embedding_dim, config.float16)?; + HDF5Memory::create(config)? + }; + let float16 = mem.config().float16; + let dim = mem.config().embedding_dim; + + let mut m = Migration { + appended_to_existing: existing, + float16, + embedding_dim: dim, + ..Migration::default() + }; + + // ---- Memory records -------------------------------------------------- + // Store indices of every record the store already holds, by content, so + // a source row that appears twice is only treated as present as often + // as the store has it. + let mut present: HashMap> = HashMap::new(); + if existing { + let c = &mem.cache; + for i in 0..c.len() { + let key = record_key( + &c.chunks[i], + &c.source_channels[i], + &c.session_ids[i], + &c.tags[i], + c.timestamps[i], + ); + present.entry(key).or_default().push(i); + } + } + let key_of = |c: &MemoryChunk| { + record_key( + &c.chunk, + &c.source_channel, + &c.session_id, + &c.tags, + c.timestamp, + ) + }; + let tombstoned = |idx: usize| mem.cache.tombstones[idx] != 0; + // Pass 1: a store record in the same deleted state as the source row. + let mut unmatched: Vec = Vec::new(); + for (i, c) in data.chunks.iter().enumerate() { + let src_deleted = c.deleted != 0; + let hit = present.get_mut(&key_of(c)).and_then(|idxs| { + let at = idxs.iter().position(|&x| tombstoned(x) == src_deleted)?; + Some(idxs.remove(at)) + }); + match hit { + Some(_) => m.chunks_present += 1, + None => unmatched.push(i), + } + } + // Pass 2: a store record whose deleted state differs — the source row + // was deleted or restored since the last migration. The source wins. + let mut new_chunks: Vec = Vec::with_capacity(unmatched.len()); + let mut delete_in_store: Vec = Vec::new(); + for i in unmatched { + let c = &data.chunks[i]; + let hit = present + .get_mut(&key_of(c)) + .and_then(|idxs| (!idxs.is_empty()).then(|| idxs.remove(0))); + match hit { + // Active in the store, deleted in the source: tombstone it. + Some(idx) if c.deleted != 0 => { + m.deleted_in_store.push((idx, i)); + delete_in_store.push(idx); + } + // Deleted in the store, active in the source. The agent has no + // un-delete, so the row is written again as a new active record + // (the tombstone stays until the store is compacted). + Some(_) => { + m.restored += 1; + new_chunks.push(i); + } + None if c.deleted != 0 && opts.skip_deleted => m.deleted_skipped += 1, + None => new_chunks.push(i), + } + } + new_chunks.sort_unstable(); + let to_write: Vec<&MemoryChunk> = new_chunks.iter().map(|&i| &data.chunks[i]).collect(); + + // ---- Sessions (in the cache; persisted by the save_batch checkpoint) --- + let known_sessions: HashSet = mem + .sessions() + .entries + .iter() + .map(|e| e.id.clone()) + .collect(); + for (i, s) in data.sessions.iter().enumerate() { + if known_sessions.contains(&s.id) { + m.sessions_present += 1; + continue; + } + let sessions = mem.sessions_mut(); + let at = sessions.len(); + sessions.add_at( + &s.id, + s.start_idx.max(0) as usize, + s.end_idx.max(0) as usize, + &s.channel, + &s.summary, + s.timestamp * US_PER_SEC, + ); + m.sessions.push((at, i)); + } + + // ---- Knowledge graph ------------------------------------------------- + let kg = mem.knowledge_mut(); + // Matched only against what the store held before this run: the source + // itself is copied as it is, duplicates included. + let by_name_type: HashMap<(String, String), u64> = kg + .entities + .iter() + .map(|e| ((e.name.clone(), e.entity_type.clone()), e.id)) + .collect(); + for (i, e) in data.entities.iter().enumerate() { + let key = (e.name.clone(), e.entity_type.clone()); + let id = match by_name_type.get(&key) { + Some(&id) => { + m.entities_present += 1; + id + } + None => { + let id = kg.add_entity(&e.name, &e.entity_type, e.embedding_idx); + m.entities.push((id, i)); + id + } + }; + m.entity_ids.insert(e.id, id); + } + let known_relations: HashSet<(u64, u64, String)> = kg + .relations + .iter() + .map(|r| (r.src, r.tgt, r.relation.clone())) + .collect(); + for (i, r) in data.relations.iter().enumerate() { + let (Some(&src), Some(&tgt)) = (m.entity_ids.get(&r.src), m.entity_ids.get(&r.tgt)) else { + m.dangling_relations.push(i); + continue; + }; + if known_relations.contains(&(src, tgt, r.relation.clone())) { + m.relations_present += 1; + continue; + } + let at = kg.relations.len(); + kg.add_relation(src, tgt, &r.relation, r.weight as f32); + kg.relations[at].ts = r.timestamp * US_PER_SEC; + m.relations.push((at, i)); + } + + // ---- Write: one checkpoint for records, sessions and graph ----------- + let entries: Vec = to_write + .iter() + .map(|c| MemoryEntry { + chunk: c.chunk.clone(), + embedding: c.embedding.clone(), + source_channel: c.source_channel.clone(), + timestamp: c.timestamp, + session_id: c.session_id.clone(), + tags: c.tags.clone(), + }) + .collect(); + let indices = mem.save_batch(entries)?; + m.records = indices + .iter() + .copied() + .zip(new_chunks.iter().copied()) + .collect(); + + // Rows deleted in the source stay deleted: tombstones, as the agent's own + // `delete` leaves them (not compacted away). + // Records matched in the store whose source row has since been deleted + // are tombstoned too. + let tombstones: Vec = m + .records + .iter() + .filter(|&&(_, src)| data.chunks[src].deleted != 0) + .map(|&(idx, _)| idx) + .chain(delete_in_store) + .collect(); + mem.delete_batch(&tombstones)?; + + m.anomaly_alerts = mem + .take_anomaly_alerts() + .into_iter() + .map(|a| a.message) + .collect(); + m.store_count = mem.count(); + drop(mem); // release the single-writer lock before anyone re-opens it + Ok(m) +} diff --git a/crates/clawhdf5-migrate/src/validate.rs b/crates/clawhdf5-migrate/src/validate.rs index 4a98d95..f028fb7 100644 --- a/crates/clawhdf5-migrate/src/validate.rs +++ b/crates/clawhdf5-migrate/src/validate.rs @@ -1,192 +1,266 @@ -use clawhdf5::reader::File as Hdf5File; -use clawhdf5_format::provenance::VerifyResult; +//! Validate a migration by reading the store back the way an agent would: +//! through `HDF5Memory::open_read_only`, comparing what it loads with the +//! SQLite source, and running a search for a migrated record. + +use std::path::Path; + +use clawhdf5_agent::{AgentMemory, HDF5Memory, SearchOptions}; +use clawhdf5_format::float16::round_to_f16; -use crate::hdf5_reader::read_hdf5; use crate::sqlite_reader::SqliteData; +use crate::store_writer::{Migration, US_PER_SEC}; type BoxErr = Box; /// Summary of a migration validation. #[derive(Debug)] pub struct ValidationSummary { - pub chunks: u64, - pub sessions: u64, - pub entities: u64, - pub relations: u64, - pub embedding_dim: u64, - /// Number of rows whose full content was compared against the source. + /// Records in the store (including tombstones). + pub count: usize, + /// Records in the store that are not deleted. + pub active: usize, + pub sessions: usize, + pub entities: usize, + pub relations: usize, + pub embedding_dim: usize, + pub float16: bool, + /// Rows whose full content was compared against the source. pub rows_checked: u64, - /// Whether the `chunks/text` and `chunks/embeddings` SHINES provenance - /// hashes (written via [`crate::hdf5_writer`]) were both present and - /// matched their recomputed SHA-256 on read-back. `false` when either - /// dataset has no provenance metadata (e.g. an older output file) or - /// there are zero chunks to check. - pub provenance_verified: bool, + /// Whether a search for a migrated record found it (`false` when there + /// was no active migrated record with an embedding to search for). + pub search_checked: bool, } -/// Validate a migrated HDF5 file against the source data. +/// Validate the store at `path` against the source rows `migration` wrote. /// -/// Reads the written file back and compares actual content — chunk text, -/// embeddings, and every session/entity/relation field — to the source, not -/// just the row counts. When `full` is false a representative sample of chunk -/// rows is content-checked (counts and all other groups are always checked in -/// full); when `full` is true every chunk row is compared too. `float16` widens -/// the embedding tolerance to allow for half-precision quantization. -pub fn validate_hdf5( - path: &str, +/// Counts and the session / entity / relation rows are always checked in +/// full. Memory records are content-checked on a representative sample, or +/// all of them with `full`. Embeddings must match exactly: the source values +/// themselves in an `f32` store, their [`round_to_f16`] in a `float16` one. +pub fn validate_store( + path: &Path, source: &SqliteData, + migration: &Migration, full: bool, - float16: bool, ) -> Result { - let got = read_hdf5(path)?; - let provenance_verified = verify_chunk_provenance(path)?; + let mut mem = HDF5Memory::open_read_only(path)?; + let float16 = mem.config().float16; + let dim = mem.config().embedding_dim; // ---- Counts ---- - check_count("chunk", got.chunks.len(), source.chunks.len())?; - check_count("session", got.sessions.len(), source.sessions.len())?; - check_count("entity", got.entities.len(), source.entities.len())?; - check_count("relation", got.relations.len(), source.relations.len())?; - if got.embedding_dim != source.embedding_dim { + check_count("record", mem.count(), migration.store_count)?; + if float16 != migration.float16 { return Err(format!( - "embedding_dim mismatch: HDF5 has {}, source has {}", - got.embedding_dim, source.embedding_dim + "float16 mismatch: store {float16}, expected {}", + migration.float16 ) .into()); } + if dim != migration.embedding_dim { + return Err(format!( + "embedding_dim mismatch: store has {dim}, expected {}", + migration.embedding_dim + ) + .into()); + } + if !migration.appended_to_existing { + check_count("record", mem.count(), migration.records.len())?; + check_count("session", mem.sessions().len(), migration.sessions.len())?; + check_count( + "entity", + mem.knowledge().entities.len(), + migration.entities.len(), + )?; + check_count( + "relation", + mem.knowledge().relations.len(), + migration.relations.len(), + )?; + } - // ---- Chunk content (sampled or full) ---- - let (emb_abs, emb_rel) = if float16 { (1e-2, 1e-2) } else { (1e-4, 0.0) }; + // ---- Memory records (sampled or full) ---- let mut rows_checked = 0u64; - for i in sample_indices(source.chunks.len(), full) { - let (s, g) = (&source.chunks[i], &got.chunks[i]); - if s.id != g.id { - return Err(field_err("chunk", i, "id", s.id, g.id)); + let expected_value = |v: f32| if float16 { round_to_f16(v) } else { v }; + for k in sample_indices(migration.records.len(), full) { + let (idx, src) = migration.records[k]; + let s = &source.chunks[src]; + let c = &mem.cache; + if idx >= c.len() { + return Err( + format!("record {idx} (chunk id {}) is missing from the store", s.id).into(), + ); } - if s.chunk != g.chunk { + let id = s.id; + if c.chunks[idx] != s.chunk { return Err(format!( - "chunk[{i}].text mismatch: source {:?}, HDF5 {:?}", + "record {idx} (chunk id {id}) text mismatch: source {:?}, store {:?}", truncate(&s.chunk), - truncate(&g.chunk) + truncate(&c.chunks[idx]) ) .into()); } - if s.session_id != g.session_id || s.source_channel != g.source_channel || s.tags != g.tags + if c.source_channels[idx] != s.source_channel + || c.session_ids[idx] != s.session_id + || c.tags[idx] != s.tags { - return Err(format!("chunk[{i}] string field mismatch").into()); + return Err(format!("record {idx} (chunk id {id}) string field mismatch").into()); } - if s.deleted != g.deleted { - return Err(field_err("chunk", i, "deleted", s.deleted, g.deleted)); - } - if s.embedding.len() != g.embedding.len() { + if c.timestamps[idx].to_bits() != s.timestamp.to_bits() { return Err(format!( - "chunk[{i}] embedding length mismatch: {} vs {}", - s.embedding.len(), - g.embedding.len() + "record {idx} (chunk id {id}) timestamp mismatch: source {}, store {}", + s.timestamp, c.timestamps[idx] ) .into()); } - for (k, (&a, &b)) in s.embedding.iter().zip(g.embedding.iter()).enumerate() { - if (a - b).abs() > emb_abs + emb_rel * a.abs() { - return Err( - format!("chunk[{i}].embedding[{k}] mismatch: source {a}, HDF5 {b}").into(), - ); + let deleted = c.tombstones[idx] != 0; + if deleted != (s.deleted != 0) { + return Err(format!( + "record {idx} (chunk id {id}) deleted mismatch: source {}, store {deleted}", + s.deleted != 0 + ) + .into()); + } + let got = c.embeddings.get(idx).unwrap_or(&[]); + if got.len() != s.embedding.len() { + return Err(format!( + "record {idx} (chunk id {id}) embedding length mismatch: source {}, store {}", + s.embedding.len(), + got.len() + ) + .into()); + } + for (j, (&a, &b)) in s.embedding.iter().zip(got).enumerate() { + let want = expected_value(a); + if want.to_bits() != b.to_bits() && !(want.is_nan() && b.is_nan()) { + return Err(format!( + "record {idx} (chunk id {id}) embedding[{j}] mismatch: source {a}, \ + expected {want}, store {b}" + ) + .into()); } } rows_checked += 1; } - // ---- Other groups (always full — they are small) ---- - for (i, (s, g)) in source.sessions.iter().zip(got.sessions.iter()).enumerate() { - if s.id != g.id - || s.start_idx != g.start_idx - || s.end_idx != g.end_idx - || s.channel != g.channel - || s.summary != g.summary - { - return Err(format!("session[{i}] mismatch").into()); + // ---- Records tombstoned because their source row was deleted ---- + for &(idx, src) in &migration.deleted_in_store { + let s = &source.chunks[src]; + let c = &mem.cache; + if idx >= c.len() || c.chunks[idx] != s.chunk || c.timestamps[idx] != s.timestamp { + return Err(format!("record {idx} (chunk id {}) mismatch or missing", s.id).into()); } - rows_checked += 1; - } - for (i, (s, g)) in source.entities.iter().zip(got.entities.iter()).enumerate() { - if s.id != g.id - || s.name != g.name - || s.entity_type != g.entity_type - || s.embedding_idx != g.embedding_idx - { - return Err(format!("entity[{i}] mismatch").into()); - } - rows_checked += 1; - } - for (i, (s, g)) in source - .relations - .iter() - .zip(got.relations.iter()) - .enumerate() - { - if s.src != g.src || s.tgt != g.tgt || s.relation != g.relation { - return Err(format!("relation[{i}] mismatch").into()); + if c.tombstones[idx] == 0 { + return Err(format!( + "record {idx} (chunk id {}) is deleted in the source but active in the store", + s.id + ) + .into()); } rows_checked += 1; } + // ---- Sessions ---- + let sessions = mem.sessions(); + for &(at, src) in &migration.sessions { + let s = &source.sessions[src]; + let (Some(e), Some(summary)) = (sessions.entries.get(at), sessions.summaries.get(at)) + else { + return Err(format!("session {:?} is missing from the store", s.id).into()); + }; + if e.id != s.id + || e.start_idx != s.start_idx.max(0) as u64 + || e.end_idx != s.end_idx.max(0) as u64 + || e.channel != s.channel + || *summary != s.summary + || e.ts != s.timestamp * US_PER_SEC + { + return Err(format!("session {:?} mismatch", s.id).into()); + } + rows_checked += 1; + } + + // ---- Knowledge graph ---- + let kg = mem.knowledge(); + for &(id, src) in &migration.entities { + let s = &source.entities[src]; + let Some(e) = kg.get_entity(id) else { + return Err(format!( + "entity {:?} (id {}) is missing from the store", + s.name, s.id + ) + .into()); + }; + if e.name != s.name || e.entity_type != s.entity_type || e.embedding_idx != s.embedding_idx + { + return Err(format!("entity {:?} (id {}) mismatch", s.name, s.id).into()); + } + rows_checked += 1; + } + for &(at, src) in &migration.relations { + let s = &source.relations[src]; + let r = kg.relations.get(at); + let ok = r.is_some_and(|r| { + Some(&r.src) == migration.entity_ids.get(&s.src) + && Some(&r.tgt) == migration.entity_ids.get(&s.tgt) + && r.relation == s.relation + && r.weight == s.weight as f32 + && r.ts == s.timestamp * US_PER_SEC + }); + if !ok { + return Err(format!( + "relation {} -[{}]-> {} mismatch or missing", + s.src, s.relation, s.tgt + ) + .into()); + } + rows_checked += 1; + } + + // ---- A migrated record must be findable by search ---- + let probe = migration + .records + .iter() + .copied() + .find(|&(idx, _)| dim > 0 && mem.cache.tombstones[idx] == 0); + let search_checked = match probe { + None => false, + Some((idx, _)) => { + let query = mem.cache.embeddings[idx].to_vec(); + let text = mem.cache.chunks[idx].clone(); + let hits = mem.search(&query, &text, &SearchOptions::new(10)); + // A record with the same text is as good a hit: the source may + // hold duplicates, and they tie. + if !hits.iter().any(|h| h.index == idx || h.chunk == text) { + return Err(format!( + "search for migrated record {idx} ({:?}) did not return it", + truncate(&text) + ) + .into()); + } + true + } + }; + Ok(ValidationSummary { - chunks: got.chunks.len() as u64, - sessions: got.sessions.len() as u64, - entities: got.entities.len() as u64, - relations: got.relations.len() as u64, - embedding_dim: got.embedding_dim as u64, + count: mem.count(), + active: mem.count_active(), + sessions: mem.sessions().len(), + entities: mem.knowledge().entities.len(), + relations: mem.knowledge().relations.len(), + embedding_dim: dim, + float16, rows_checked, - provenance_verified, + search_checked, }) } fn check_count(kind: &str, got: usize, expected: usize) -> Result<(), BoxErr> { if got != expected { - return Err(format!("{kind} count mismatch: HDF5 has {got}, source has {expected}").into()); + return Err(format!("{kind} count mismatch: store has {got}, expected {expected}").into()); } Ok(()) } -/// Re-verify the SHA-256 provenance hash of `chunks/text` and -/// `chunks/embeddings` against their actual stored bytes, catching -/// post-write corruption that a plain content comparison against the -/// in-memory source wouldn't (the source is compared against what -/// `read_hdf5` decoded, not against the raw bytes on disk). -/// -/// Returns `Ok(true)` only if both datasets exist and both hashes match. -/// Returns `Ok(false)` (not an error) if a dataset has no provenance -/// attributes at all (e.g. a file written before this check existed) or -/// there are zero chunks. Returns an error only on an actual hash mismatch — -/// that indicates real corruption. -fn verify_chunk_provenance(path: &str) -> Result { - let file = Hdf5File::open(path)?; - let Ok(chunks) = file.group("chunks") else { - return Ok(false); - }; - let mut all_present = true; - for name in ["text", "embeddings"] { - let Ok(ds) = chunks.dataset(name) else { - all_present = false; - continue; - }; - match ds.verify_provenance()? { - VerifyResult::Ok => {} - VerifyResult::NoHash => all_present = false, - VerifyResult::Mismatch { stored, computed } => { - return Err(format!( - "provenance hash mismatch on chunks/{name}: stored {stored}, recomputed {computed} — data may be corrupted" - ) - .into()); - } - } - } - Ok(all_present) -} - -fn field_err(kind: &str, i: usize, field: &str, s: T, g: T) -> BoxErr { - format!("{kind}[{i}].{field} mismatch: source {s}, HDF5 {g}").into() -} - fn truncate(s: &str) -> String { if s.len() <= 40 { s.to_string() @@ -196,7 +270,7 @@ fn truncate(s: &str) -> String { } } -/// Indices of chunk rows to content-check. Full = all; otherwise a spread of +/// Indices of records to content-check. Full = all; otherwise a spread of /// representative rows (first/last and evenly-spaced interior samples). fn sample_indices(n: usize, full: bool) -> Vec { if n == 0 {