diff --git a/Cargo.lock b/Cargo.lock index 4eec046f2..4f779bf44 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2824,6 +2824,18 @@ version = "0.32.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e629b9b98ef3dd8afe6ca2bd0f89306cec16d43d907889945bc5d6687f2f13c7" +[[package]] +name = "git2" +version = "0.21.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ddddbf932745a6be37109b6112d3ee09696106f848449069d3a57bba937ab82e" +dependencies = [ + "bitflags 2.11.1", + "libc", + "libgit2-sys", + "log", +] + [[package]] name = "glob" version = "0.3.3" @@ -3862,6 +3874,18 @@ version = "0.2.186" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "68ab91017fe16c622486840e4c83c9a37afeff978bd239b5293d61ece587de66" +[[package]] +name = "libgit2-sys" +version = "0.18.5+1.9.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "005d6ae6eac1912906073e069f7db60b1fa98e052a68227824afe3e3a1c59ca2" +dependencies = [ + "cc", + "libc", + "libz-sys", + "pkg-config", +] + [[package]] name = "libloading" version = "0.8.9" @@ -3898,6 +3922,18 @@ dependencies = [ "vcpkg", ] +[[package]] +name = "libz-sys" +version = "1.1.29" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "85bc9657773828b90eeb625adff10eeac83cc21bbfd8e23a03eaa8a33c9e28d9" +dependencies = [ + "cc", + "libc", + "pkg-config", + "vcpkg", +] + [[package]] name = "linux-keyutils" version = "0.2.5" @@ -5169,6 +5205,7 @@ dependencies = [ "fs2", "futures", "futures-util", + "git2", "glob", "hex", "hkdf", diff --git a/Cargo.toml b/Cargo.toml index 70f0077b2..acbe63871 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -70,6 +70,10 @@ dirs = "5" sha2 = "0.10" # Line-level text diffs for the memory_diff module (modified-item unified diffs). similar = "2" +# Git-backed change ledger for the memory_diff module: snapshots are commits, +# checkpoints are tags, read markers are refs, diffs are git tree diffs. +# Vendored libgit2 (no system git dependency on end-user machines). +git2 = { version = "0.21", default-features = false, features = ["vendored-libgit2"] } # Legacy SHA-1 only used for Tencent COS HMAC-SHA1 signing (yuanbao # channel media upload). Not used for any new security-sensitive work. sha1 = "0.10" diff --git a/app/src-tauri/Cargo.lock b/app/src-tauri/Cargo.lock index 8847e885e..0fc817abe 100644 --- a/app/src-tauri/Cargo.lock +++ b/app/src-tauri/Cargo.lock @@ -3205,6 +3205,18 @@ dependencies = [ "winapi", ] +[[package]] +name = "git2" +version = "0.21.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ddddbf932745a6be37109b6112d3ee09696106f848449069d3a57bba937ab82e" +dependencies = [ + "bitflags 2.11.1", + "libc", + "libgit2-sys", + "log", +] + [[package]] name = "glib" version = "0.18.5" @@ -4302,6 +4314,18 @@ dependencies = [ "pkg-config", ] +[[package]] +name = "libgit2-sys" +version = "0.18.5+1.9.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "005d6ae6eac1912906073e069f7db60b1fa98e052a68227824afe3e3a1c59ca2" +dependencies = [ + "cc", + "libc", + "libz-sys", + "pkg-config", +] + [[package]] name = "libloading" version = "0.7.4" @@ -4358,6 +4382,18 @@ dependencies = [ "vcpkg", ] +[[package]] +name = "libz-sys" +version = "1.1.29" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "85bc9657773828b90eeb625adff10eeac83cc21bbfd8e23a03eaa8a33c9e28d9" +dependencies = [ + "cc", + "libc", + "pkg-config", + "vcpkg", +] + [[package]] name = "linux-keyutils" version = "0.2.5" @@ -5428,6 +5464,7 @@ dependencies = [ "fs2", "futures", "futures-util", + "git2", "glob", "hex", "hkdf", diff --git a/src/openhuman/memory_diff/git_store.rs b/src/openhuman/memory_diff/git_store.rs new file mode 100644 index 000000000..86b471ee2 --- /dev/null +++ b/src/openhuman/memory_diff/git_store.rs @@ -0,0 +1,868 @@ +//! Git-backed persistence for memory diff snapshots, checkpoints, and read +//! markers. +//! +//! The ledger is a libgit2 repository at `/memory_diff/repo`. +//! It is a *derived* view of the chunk store — `mem_tree_chunks` remains the +//! authoritative source of memory. Each snapshot materialises a source's +//! current items as blobs under `/` and records them as a commit; +//! the rest of the tree (other sources) is carried forward from the parent so +//! HEAD always reflects the whole world. This maps the diff domain onto git's +//! native primitives: +//! +//! - **Snapshot** → commit (`Snapshot.id` is the commit SHA) +//! - **Checkpoint**→ annotated tag `ckpt_` at HEAD +//! - **Read marker**→ ref `refs/openhuman/read/` → commit SHA +//! - **Diff** → `git diff ..` scoped to the source path +//! +//! Item identity is the file name: each item is one flat blob whose name is the +//! item id encoded into a git-safe path component (`encode_item_id`). A content +//! change keeps the same name → `Modified`; renaming the item id is +//! `Removed` + `Added`, matching the previous id-keyed semantics. +//! +//! Snapshot metadata that has no natural git home (source kind/label, trigger, +//! item count, millisecond timestamp) rides in the commit message as trailers. +//! All mutations serialise through a process-global [`WRITE_LOCK`] because the +//! repository's parent/HEAD bookkeeping is not safe to interleave. + +use std::collections::HashMap; +use std::path::Path; +use std::sync::Mutex; + +use anyhow::{Context, Result}; +use git2::{Delta, DiffOptions, Object, ObjectType, Oid, Repository, Signature, Time}; + +use super::types::{ChangeKind, Checkpoint, DiffSummary, ItemChange, Snapshot, SnapshotTrigger}; + +/// Serialises all writes (commits, tags, ref updates) to the ledger. libgit2's +/// HEAD/parent resolution is read-modify-write, so concurrent commits could +/// otherwise fork history or lose a snapshot. +static WRITE_LOCK: Mutex<()> = Mutex::new(()); + +const BLOB_MODE: i32 = 0o100644; +const TREE_MODE: i32 = 0o040000; +const SIG_NAME: &str = "OpenHuman Memory"; +const SIG_EMAIL: &str = "memory-diff@openhuman.local"; +const READ_MARKER_PREFIX: &str = "refs/openhuman/read/"; +const CHECKPOINT_PREFIX: &str = "ckpt_"; + +/// Upper bound on a single modified-item unified diff embedded in `text_diff`. +const MAX_TEXT_DIFF_CHARS: usize = 2000; + +// ── Repository handle ────────────────────────────────────────────────── + +/// A handle to the diff ledger. Cheap to open; callers construct one per +/// blocking task (mirroring the previous `with_connection` pattern). +pub struct Ledger { + repo: Repository, +} + +/// Metadata describing the snapshot a commit represents. Persisted as commit +/// trailers and reconstructed by [`Ledger::snapshot_from_commit`]. +pub struct SnapshotMeta { + pub source_id: String, + pub source_kind: String, + pub label: String, + pub trigger: SnapshotTrigger, +} + +impl Ledger { + /// Open the ledger, initialising the repository on first use. + pub fn open(workspace_dir: &Path) -> Result { + let repo_path = workspace_dir.join("memory_diff").join("repo"); + std::fs::create_dir_all(&repo_path) + .with_context(|| format!("create memory_diff repo dir: {}", repo_path.display()))?; + + let repo = match Repository::open(&repo_path) { + Ok(repo) => repo, + Err(_) => { + tracing::debug!( + path = %repo_path.display(), + "[memory_diff::git] initialising diff ledger repository" + ); + Repository::init(&repo_path) + .with_context(|| format!("init memory_diff repo: {}", repo_path.display()))? + } + }; + Ok(Self { repo }) + } + + // ── Snapshots (commits) ──────────────────────────────────────────── + + /// Commit a snapshot for one source: replace the source's subtree with the + /// given items (each `(item_id, content)`), carrying every other source + /// forward from the parent. Returns the resulting [`Snapshot`]. + pub fn commit_snapshot( + &self, + meta: &SnapshotMeta, + items: &[(String, String)], + taken_at_ms: i64, + ) -> Result { + let _guard = WRITE_LOCK.lock().expect("memory_diff write lock poisoned"); + + // Build the source subtree from scratch: one blob per item. + let source_tree_oid = { + let mut tb = self.repo.treebuilder(None)?; + for (item_id, content) in items { + let blob = self.repo.blob(content.as_bytes())?; + tb.insert(encode_item_id(item_id), blob, BLOB_MODE)?; + } + tb.write()? + }; + + // Start the root tree from the parent commit (carry other sources), + // then graft in the new source subtree (or drop it if empty). + let parent_commit = match self.repo.head() { + Ok(head) => Some(head.peel_to_commit()?), + Err(_) => None, // unborn HEAD on a fresh repo + }; + let parent_root = match &parent_commit { + Some(c) => Some(c.tree()?), + None => None, + }; + let root_oid = { + let mut tb = self.repo.treebuilder(parent_root.as_ref())?; + if items.is_empty() { + if tb.get(meta.source_id.as_str())?.is_some() { + tb.remove(meta.source_id.as_str())?; + } + } else { + tb.insert(meta.source_id.as_str(), source_tree_oid, TREE_MODE)?; + } + tb.write()? + }; + let tree = self.repo.find_tree(root_oid)?; + + let message = build_commit_message(meta, items.len() as u32, taken_at_ms); + let sig = signature(taken_at_ms)?; + let parents: Vec<&git2::Commit> = parent_commit.iter().collect(); + let commit_oid = self + .repo + .commit(Some("HEAD"), &sig, &sig, &message, &tree, &parents) + .context("write snapshot commit")?; + + tracing::debug!( + commit = %commit_oid, + source_id = %meta.source_id, + items = items.len(), + "[memory_diff::git] snapshot committed" + ); + + Ok(Snapshot { + id: commit_oid.to_string(), + source_id: meta.source_id.clone(), + source_kind: meta.source_kind.clone(), + label: meta.label.clone(), + trigger: meta.trigger.clone(), + item_count: items.len() as u32, + taken_at_ms, + }) + } + + /// List snapshots newest-first, optionally filtered to one source. + /// + /// Walks the commit history from HEAD; each commit is one source's + /// snapshot, identified by its `Source-Id` trailer. + pub fn list_snapshots(&self, source_id: Option<&str>, limit: u32) -> Result> { + let mut walk = match self.repo.revwalk() { + Ok(w) => w, + Err(_) => return Ok(Vec::new()), + }; + if walk.push_head().is_err() { + // Unborn HEAD → no snapshots yet. + return Ok(Vec::new()); + } + walk.set_sorting(git2::Sort::TIME)?; + + let mut out = Vec::new(); + for oid in walk { + let oid = oid?; + let commit = self.repo.find_commit(oid)?; + let snap = self.snapshot_from_commit(&commit); + if let Some(filter) = source_id { + if snap.source_id != filter { + continue; + } + } + out.push(snap); + if out.len() as u32 >= limit { + break; + } + } + Ok(out) + } + + /// Fetch a single snapshot by commit SHA, if it exists. + pub fn get_snapshot(&self, snapshot_id: &str) -> Result> { + let Ok(oid) = Oid::from_str(snapshot_id) else { + return Ok(None); + }; + match self.repo.find_commit(oid) { + Ok(commit) => Ok(Some(self.snapshot_from_commit(&commit))), + Err(_) => Ok(None), + } + } + + /// The `count` most recent snapshots for a source, newest-first. + pub fn latest_snapshots_for_source( + &self, + source_id: &str, + count: u32, + ) -> Result> { + self.list_snapshots(Some(source_id), count) + } + + /// Number of distinct sources that have at least one snapshot. + pub fn snapshot_count_for_source(&self, source_id: &str) -> Result { + Ok(self.list_snapshots(Some(source_id), u32::MAX)?.len()) + } + + // ── Diff (tree-to-tree) ───────────────────────────────────────────── + + /// Compute item-level changes for `source_id` between two snapshots. + /// + /// `from` is `None` for a first-ever diff (everything added). Both commits + /// must belong to `source_id`; cross-source mixing is rejected by the + /// caller before reaching here. + pub fn compute_changes( + &self, + from: Option<&str>, + to: &str, + source_id: &str, + to_item_count: u32, + include_text_diff: bool, + ) -> Result<(Vec, DiffSummary)> { + let to_oid = Oid::from_str(to).with_context(|| format!("bad to snapshot id: {to}"))?; + let to_tree = self.repo.find_commit(to_oid)?.tree()?; + + let from_tree = match from { + Some(f) => { + let oid = Oid::from_str(f).with_context(|| format!("bad from snapshot id: {f}"))?; + Some(self.repo.find_commit(oid)?.tree()?) + } + None => None, + }; + + let path_prefix = format!("{source_id}/"); + let mut opts = DiffOptions::new(); + opts.pathspec(source_id); + opts.context_lines(3); + let diff = + self.repo + .diff_tree_to_tree(from_tree.as_ref(), Some(&to_tree), Some(&mut opts))?; + + let mut changes = Vec::new(); + let mut summary = DiffSummary::default(); + + for (idx, delta) in diff.deltas().enumerate() { + // Resolve the item path; guard against pathspec prefix overreach + // (e.g. "src_a" must not match "src_abc/..."). + let path = delta + .new_file() + .path() + .or_else(|| delta.old_file().path()) + .and_then(|p| p.to_str()) + .unwrap_or(""); + let Some(encoded) = path.strip_prefix(&path_prefix) else { + continue; + }; + let item_id = decode_item_id(encoded); + + let new_oid = delta.new_file().id(); + let old_oid = delta.old_file().id(); + + let (kind, title) = match delta.status() { + Delta::Added | Delta::Copied | Delta::Untracked => { + summary.added += 1; + (ChangeKind::Added, self.title_for(&item_id, new_oid)) + } + Delta::Deleted => { + summary.removed += 1; + (ChangeKind::Removed, self.title_for(&item_id, old_oid)) + } + Delta::Modified | Delta::Renamed | Delta::Typechange => { + summary.modified += 1; + (ChangeKind::Modified, self.title_for(&item_id, new_oid)) + } + // Unmodified / ignored / conflicted: nothing to report. + _ => continue, + }; + + let text_diff = if include_text_diff && kind == ChangeKind::Modified { + patch_text(&diff, idx) + } else { + None + }; + + changes.push(ItemChange { + item_id, + title, + kind, + old_content_hash: oid_hash(old_oid), + new_content_hash: oid_hash(new_oid), + text_diff, + }); + } + + // git only reports changed entries; unchanged = everything in `to` + // that wasn't added or modified. + summary.unchanged = to_item_count + .saturating_sub(summary.added) + .saturating_sub(summary.modified); + + Ok((changes, summary)) + } + + // ── Read markers (refs) ───────────────────────────────────────────── + + /// The commit SHA a source's read marker points at, if set. + pub fn get_read_marker(&self, source_id: &str) -> Result> { + let name = read_marker_ref(source_id); + match self.repo.find_reference(&name) { + Ok(r) => Ok(r.target().map(|o| o.to_string())), + Err(_) => Ok(None), + } + } + + /// Set (or advance) a source's read marker to a commit SHA. + pub fn set_read_marker(&self, source_id: &str, snapshot_id: &str) -> Result<()> { + let _guard = WRITE_LOCK.lock().expect("memory_diff write lock poisoned"); + let oid = Oid::from_str(snapshot_id) + .with_context(|| format!("bad read-marker snapshot id: {snapshot_id}"))?; + let name = read_marker_ref(source_id); + self.repo + .reference(&name, oid, true, "advance memory_diff read marker") + .with_context(|| format!("set read marker ref: {name}"))?; + Ok(()) + } + + // ── Checkpoints (tags) ────────────────────────────────────────────── + + /// Create an annotated tag at HEAD recording a checkpoint. The label and + /// per-source head snapshot ids ride in the tag message as JSON. + pub fn create_checkpoint( + &self, + id: &str, + label: &str, + snapshot_ids: &[String], + created_at_ms: i64, + ) -> Result<()> { + let _guard = WRITE_LOCK.lock().expect("memory_diff write lock poisoned"); + let head = self + .repo + .head() + .context("checkpoint requires at least one snapshot")? + .peel_to_commit()?; + let target: Object = head.into_object(); + let sig = signature(created_at_ms)?; + let message = checkpoint_message(label, snapshot_ids, created_at_ms); + self.repo + .tag(id, &target, &sig, &message, true) + .with_context(|| format!("create checkpoint tag: {id}"))?; + Ok(()) + } + + /// Load a checkpoint by tag name. + pub fn get_checkpoint(&self, checkpoint_id: &str) -> Result> { + let refname = format!("refs/tags/{checkpoint_id}"); + let Ok(reference) = self.repo.find_reference(&refname) else { + return Ok(None); + }; + let obj = reference.peel(ObjectType::Tag).ok(); + let Some(tag) = obj.and_then(|o| o.into_tag().ok()) else { + return Ok(None); + }; + Ok(Some(checkpoint_from_message( + checkpoint_id, + tag.message().ok().flatten().unwrap_or(""), + ))) + } + + /// List checkpoints newest-first, up to `limit`. + pub fn list_checkpoints(&self, limit: u32) -> Result> { + let pattern = format!("{CHECKPOINT_PREFIX}*"); + let names = self.repo.tag_names(Some(&pattern))?; + let mut out = Vec::new(); + // StringArray::iter() yields Result, _>; drop errors/non-utf8. + for name in names.iter().flatten().flatten() { + if let Some(ckpt) = self.get_checkpoint(name)? { + out.push(ckpt); + } + } + out.sort_by(|a, b| b.created_at_ms.cmp(&a.created_at_ms)); + out.truncate(limit as usize); + Ok(out) + } + + /// Delete checkpoint tags created before `older_than_ms`. Snapshot commits + /// are retained — git history is the ledger — so this only prunes named + /// baselines. Returns the number of tags deleted. + pub fn cleanup_checkpoints(&self, older_than_ms: i64) -> Result { + let _guard = WRITE_LOCK.lock().expect("memory_diff write lock poisoned"); + let pattern = format!("{CHECKPOINT_PREFIX}*"); + let names = self.repo.tag_names(Some(&pattern))?; + let mut deleted = 0u64; + for name in names.iter().flatten().flatten() { + if let Some(ckpt) = self.get_checkpoint(name)? { + if ckpt.created_at_ms < older_than_ms { + self.repo.tag_delete(name)?; + deleted += 1; + } + } + } + Ok(deleted) + } + + // ── Helpers ───────────────────────────────────────────────────────── + + /// Reconstruct a [`Snapshot`] from a commit's trailers, falling back to + /// the commit time when a millisecond trailer is absent. + fn snapshot_from_commit(&self, commit: &git2::Commit) -> Snapshot { + let trailers = parse_trailers(commit.message().unwrap_or("")); + let taken_at_ms = trailers + .get("taken-at-ms") + .and_then(|s| s.parse::().ok()) + .unwrap_or_else(|| commit.time().seconds() * 1000); + Snapshot { + id: commit.id().to_string(), + source_id: trailers.get("source-id").cloned().unwrap_or_default(), + source_kind: trailers.get("source-kind").cloned().unwrap_or_default(), + label: trailers.get("source-label").cloned().unwrap_or_default(), + trigger: match trailers.get("trigger").map(String::as_str) { + Some("manual") => SnapshotTrigger::Manual, + _ => SnapshotTrigger::Auto, + }, + item_count: trailers + .get("item-count") + .and_then(|s| s.parse::().ok()) + .unwrap_or(0), + taken_at_ms, + } + } + + /// Derive a display title from a blob's content. Returns the item id when + /// the blob is missing or yields no usable line. + fn title_for(&self, item_id: &str, oid: Oid) -> String { + if oid.is_zero() { + return item_id.to_string(); + } + match self.repo.find_blob(oid) { + Ok(blob) => { + let content = String::from_utf8_lossy(blob.content()); + derive_title(item_id, &content) + } + Err(_) => item_id.to_string(), + } + } +} + +// ── Free helpers ─────────────────────────────────────────────────────── + +fn signature(at_ms: i64) -> Result> { + let time = Time::new(at_ms / 1000, 0); + Signature::new(SIG_NAME, SIG_EMAIL, &time).context("build git signature") +} + +fn read_marker_ref(source_id: &str) -> String { + format!("{READ_MARKER_PREFIX}{}", encode_item_id(source_id)) +} + +fn build_commit_message(meta: &SnapshotMeta, item_count: u32, taken_at_ms: i64) -> String { + format!( + "snapshot: {source} ({count} item(s))\n\n\ + Source-Id: {source}\n\ + Source-Kind: {kind}\n\ + Source-Label: {label}\n\ + Trigger: {trigger}\n\ + Item-Count: {count}\n\ + Taken-At-Ms: {taken}\n", + source = meta.source_id, + kind = meta.source_kind, + label = sanitize_trailer(&meta.label), + trigger = meta.trigger.as_str(), + count = item_count, + taken = taken_at_ms, + ) +} + +/// Trailer values are single-line; collapse newlines so a multi-line label +/// can't corrupt the trailer block. +fn sanitize_trailer(s: &str) -> String { + s.replace(['\n', '\r'], " ") +} + +/// Parse `Key: value` trailer lines from a commit message into a lowercase-keyed map. +fn parse_trailers(message: &str) -> HashMap { + let mut map = HashMap::new(); + for line in message.lines() { + if let Some((k, v)) = line.split_once(':') { + let key = k.trim().to_ascii_lowercase(); + if !key.is_empty() && !key.contains(' ') { + map.insert(key, v.trim().to_string()); + } + } + } + map +} + +fn checkpoint_message(label: &str, snapshot_ids: &[String], created_at_ms: i64) -> String { + let payload = serde_json::json!({ + "label": label, + "snapshot_ids": snapshot_ids, + "created_at_ms": created_at_ms, + }); + payload.to_string() +} + +fn checkpoint_from_message(id: &str, message: &str) -> Checkpoint { + let value: serde_json::Value = serde_json::from_str(message.trim()).unwrap_or_default(); + Checkpoint { + id: id.to_string(), + label: value + .get("label") + .and_then(|v| v.as_str()) + .unwrap_or("") + .to_string(), + created_at_ms: value + .get("created_at_ms") + .and_then(|v| v.as_i64()) + .unwrap_or(0), + snapshot_ids: value + .get("snapshot_ids") + .and_then(|v| v.as_array()) + .map(|arr| { + arr.iter() + .filter_map(|v| v.as_str().map(str::to_string)) + .collect() + }) + .unwrap_or_default(), + } +} + +/// A git blob oid as a content hash, or `None` for the zero oid (absent side). +fn oid_hash(oid: Oid) -> Option { + if oid.is_zero() { + None + } else { + Some(oid.to_string()) + } +} + +/// Render a single delta's unified patch, truncated to [`MAX_TEXT_DIFF_CHARS`]. +fn patch_text(diff: &git2::Diff, delta_idx: usize) -> Option { + let mut patch = git2::Patch::from_diff(diff, delta_idx).ok().flatten()?; + let buf = patch.to_buf().ok()?; + let text = buf.as_str().ok()?; + if text.trim().is_empty() { + None + } else { + Some(truncate(text, MAX_TEXT_DIFF_CHARS)) + } +} + +/// Encode an item id into a single git-safe path component. Bytes outside +/// `[A-Za-z0-9._-]` become `%XX`; an `i_` prefix keeps the result clear of the +/// reserved names `.`/`..`/empty. Reversible via [`decode_item_id`]. +fn encode_item_id(item_id: &str) -> String { + let mut out = String::with_capacity(item_id.len() + 2); + out.push_str("i_"); + for &b in item_id.as_bytes() { + if b.is_ascii_alphanumeric() || matches!(b, b'.' | b'_' | b'-') { + out.push(b as char); + } else { + out.push('%'); + out.push_str(&format!("{b:02X}")); + } + } + out +} + +/// Inverse of [`encode_item_id`]. +fn decode_item_id(encoded: &str) -> String { + let body = encoded.strip_prefix("i_").unwrap_or(encoded); + let bytes = body.as_bytes(); + let mut out = Vec::with_capacity(bytes.len()); + let mut i = 0; + while i < bytes.len() { + if bytes[i] == b'%' && i + 2 < bytes.len() { + let hex = std::str::from_utf8(&bytes[i + 1..i + 3]).unwrap_or(""); + if let Ok(byte) = u8::from_str_radix(hex, 16) { + out.push(byte); + i += 3; + continue; + } + } + out.push(bytes[i]); + i += 1; + } + String::from_utf8_lossy(&out).into_owned() +} + +fn truncate(s: &str, max_chars: usize) -> String { + if s.len() <= max_chars { + s.to_string() + } else { + let mut end = max_chars; + while !s.is_char_boundary(end) && end > 0 { + end -= 1; + } + format!("{}…(truncated)", &s[..end]) + } +} + +/// Derive a human-readable title from item content: the first non-empty line +/// (Markdown heading markers stripped), bounded. Falls back to the item id. +fn derive_title(item_id: &str, content: &str) -> String { + let first_line = content + .lines() + .map(str::trim) + .find(|l| !l.is_empty()) + .map(|l| l.trim_start_matches('#').trim()); + match first_line { + Some(l) if !l.is_empty() => truncate(l, 120), + _ => item_id.to_string(), + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn temp_ledger() -> (Ledger, tempfile::TempDir) { + let dir = tempfile::tempdir().unwrap(); + let ledger = Ledger::open(dir.path()).unwrap(); + (ledger, dir) + } + + fn meta(source_id: &str) -> SnapshotMeta { + SnapshotMeta { + source_id: source_id.to_string(), + source_kind: "folder".to_string(), + label: "Docs".to_string(), + trigger: SnapshotTrigger::Auto, + } + } + + fn items(pairs: &[(&str, &str)]) -> Vec<(String, String)> { + pairs + .iter() + .map(|(k, v)| (k.to_string(), v.to_string())) + .collect() + } + + #[test] + fn encode_decode_round_trips() { + for id in [ + "readme.md", + "path/to/file.md", + "user@example.com:msg_xxx", + "weird name (1)!", + "..", + ".", + ] { + let enc = encode_item_id(id); + assert!(!enc.contains('/'), "no slash in {enc}"); + assert!(enc != "." && enc != ".." && !enc.is_empty()); + assert_eq!(decode_item_id(&enc), id, "round trip for {id}"); + } + } + + #[test] + fn commit_and_list_snapshots() { + let (ledger, _dir) = temp_ledger(); + assert!(ledger.list_snapshots(None, 10).unwrap().is_empty()); + + let snap = ledger + .commit_snapshot(&meta("src_a"), &items(&[("a", "alpha")]), 1000) + .unwrap(); + assert_eq!(snap.source_id, "src_a"); + assert_eq!(snap.item_count, 1); + assert_eq!(snap.taken_at_ms, 1000); + + let listed = ledger.list_snapshots(Some("src_a"), 10).unwrap(); + assert_eq!(listed.len(), 1); + assert_eq!(listed[0].id, snap.id); + + let fetched = ledger.get_snapshot(&snap.id).unwrap().unwrap(); + assert_eq!(fetched.source_id, "src_a"); + assert_eq!(fetched.label, "Docs"); + assert_eq!(fetched.item_count, 1); + } + + #[test] + fn snapshots_carry_other_sources_forward() { + let (ledger, _dir) = temp_ledger(); + ledger + .commit_snapshot(&meta("src_a"), &items(&[("a", "alpha")]), 1000) + .unwrap(); + let b = ledger + .commit_snapshot(&meta("src_b"), &items(&[("b", "beta")]), 2000) + .unwrap(); + + // src_a remains listable after a src_b commit (carried forward in tree). + assert_eq!(ledger.list_snapshots(Some("src_a"), 10).unwrap().len(), 1); + assert_eq!(ledger.list_snapshots(Some("src_b"), 10).unwrap().len(), 1); + assert_eq!(ledger.list_snapshots(None, 10).unwrap().len(), 2); + assert_eq!(b.source_id, "src_b"); + } + + #[test] + fn compute_changes_added_modified_removed_unchanged() { + let (ledger, _dir) = temp_ledger(); + let from = ledger + .commit_snapshot( + &meta("src_a"), + &items(&[("a", "alpha"), ("b", "beta"), ("c", "gamma")]), + 1000, + ) + .unwrap(); + let to = ledger + .commit_snapshot( + &meta("src_a"), + &items(&[("a", "alpha"), ("b", "beta v2"), ("d", "delta")]), + 2000, + ) + .unwrap(); + + let (changes, summary) = ledger + .compute_changes(Some(&from.id), &to.id, "src_a", 3, false) + .unwrap(); + assert_eq!(summary.added, 1, "d added"); + assert_eq!(summary.modified, 1, "b modified"); + assert_eq!(summary.removed, 1, "c removed"); + assert_eq!(summary.unchanged, 1, "a unchanged"); + + let kind_of = |id: &str| { + changes + .iter() + .find(|c| c.item_id == id) + .map(|c| c.kind.clone()) + }; + assert_eq!(kind_of("d"), Some(ChangeKind::Added)); + assert_eq!(kind_of("b"), Some(ChangeKind::Modified)); + assert_eq!(kind_of("c"), Some(ChangeKind::Removed)); + assert_eq!(kind_of("a"), None); + } + + #[test] + fn compute_changes_from_none_marks_all_added() { + let (ledger, _dir) = temp_ledger(); + let to = ledger + .commit_snapshot(&meta("src_a"), &items(&[("a", "x")]), 1000) + .unwrap(); + let (changes, summary) = ledger + .compute_changes(None, &to.id, "src_a", 1, false) + .unwrap(); + assert_eq!(summary.added, 1); + assert_eq!(changes.len(), 1); + } + + #[test] + fn compute_changes_text_diff_only_when_requested() { + let (ledger, _dir) = temp_ledger(); + let from = ledger + .commit_snapshot( + &meta("src_a"), + &items(&[("a", "line one\nline two\n")]), + 1000, + ) + .unwrap(); + let to = ledger + .commit_snapshot( + &meta("src_a"), + &items(&[("a", "line one\nline TWO changed\n")]), + 2000, + ) + .unwrap(); + + let (without, _) = ledger + .compute_changes(Some(&from.id), &to.id, "src_a", 1, false) + .unwrap(); + assert!(without[0].text_diff.is_none()); + + let (with, _) = ledger + .compute_changes(Some(&from.id), &to.id, "src_a", 1, true) + .unwrap(); + let td = with[0].text_diff.as_ref().expect("text diff present"); + assert!(td.contains("line TWO changed"), "got: {td}"); + } + + #[test] + fn pathspec_does_not_leak_across_prefixed_sources() { + let (ledger, _dir) = temp_ledger(); + ledger + .commit_snapshot(&meta("src_a"), &items(&[("a", "x")]), 1000) + .unwrap(); + // src_abc shares the "src_a" prefix; its items must not appear in + // src_a's diff. + let abc = ledger + .commit_snapshot(&meta("src_abc"), &items(&[("z", "zeta")]), 2000) + .unwrap(); + let a2 = ledger + .commit_snapshot(&meta("src_a"), &items(&[("a", "x"), ("b", "y")]), 3000) + .unwrap(); + + let (changes, summary) = ledger + .compute_changes(Some(&abc.id), &a2.id, "src_a", 2, false) + .unwrap(); + assert_eq!(summary.added, 1, "only b is new in src_a"); + assert!(changes.iter().all(|c| c.item_id != "z")); + } + + #[test] + fn read_marker_set_and_get() { + let (ledger, _dir) = temp_ledger(); + assert_eq!(ledger.get_read_marker("src_a").unwrap(), None); + let snap = ledger + .commit_snapshot(&meta("src_a"), &items(&[("a", "x")]), 1000) + .unwrap(); + ledger.set_read_marker("src_a", &snap.id).unwrap(); + assert_eq!( + ledger.get_read_marker("src_a").unwrap().as_deref(), + Some(snap.id.as_str()) + ); + assert_eq!(ledger.get_read_marker("src_b").unwrap(), None); + } + + #[test] + fn checkpoint_round_trip() { + let (ledger, _dir) = temp_ledger(); + let a = ledger + .commit_snapshot(&meta("src_a"), &items(&[("a", "x")]), 1000) + .unwrap(); + let b = ledger + .commit_snapshot(&meta("src_b"), &items(&[("b", "y")]), 1000) + .unwrap(); + ledger + .create_checkpoint("ckpt_1", "baseline", &[a.id.clone(), b.id.clone()], 1500) + .unwrap(); + + let loaded = ledger.get_checkpoint("ckpt_1").unwrap().unwrap(); + assert_eq!(loaded.label, "baseline"); + assert_eq!(loaded.created_at_ms, 1500); + assert_eq!(loaded.snapshot_ids.len(), 2); + + let all = ledger.list_checkpoints(10).unwrap(); + assert_eq!(all.len(), 1); + assert_eq!(all[0].id, "ckpt_1"); + } + + #[test] + fn cleanup_checkpoints_removes_old_tags() { + let (ledger, _dir) = temp_ledger(); + ledger + .commit_snapshot(&meta("src_a"), &items(&[("a", "x")]), 1000) + .unwrap(); + ledger + .create_checkpoint("ckpt_old", "old", &[], 100) + .unwrap(); + ledger + .create_checkpoint("ckpt_new", "new", &[], 5000) + .unwrap(); + + let deleted = ledger.cleanup_checkpoints(1000).unwrap(); + assert_eq!(deleted, 1); + let remaining = ledger.list_checkpoints(10).unwrap(); + assert_eq!(remaining.len(), 1); + assert_eq!(remaining[0].id, "ckpt_new"); + } +} diff --git a/src/openhuman/memory_diff/mod.rs b/src/openhuman/memory_diff/mod.rs index 46cdf2102..827febbe6 100644 --- a/src/openhuman/memory_diff/mod.rs +++ b/src/openhuman/memory_diff/mod.rs @@ -8,16 +8,22 @@ //! Snapshots are built from already-ingested data in `mem_tree_chunks` //! (not by re-calling source readers), making them free of API calls. //! +//! Storage is a git repository at `/memory_diff/repo` (the diff +//! *ledger*): snapshots are commits, checkpoints are tags, read markers are +//! refs, and diffs are git tree diffs. `mem_tree_chunks` stays authoritative; +//! the ledger is a derived view used purely for change tracking. See +//! [`git_store`] for the mapping. +//! //! Features: //! - Per-source snapshots (auto after sync, or manual via RPC) //! - Diff between any two snapshots //! - Named checkpoints for cross-source "what changed since X" queries //! - Agent tool for in-conversation diff queries +pub mod git_store; pub mod ops; pub mod rpc; pub mod schemas; -pub mod store; pub mod tools; pub mod types; @@ -28,5 +34,5 @@ pub use schemas::{ pub use tools::MemoryDiffTool; pub use types::{ ChangeKind, Checkpoint, CrossSourceDiff, DiffResult, DiffSummary, ItemChange, Snapshot, - SnapshotItem, SnapshotTrigger, + SnapshotTrigger, }; diff --git a/src/openhuman/memory_diff/ops.rs b/src/openhuman/memory_diff/ops.rs index 040891ea9..c4c9ddde9 100644 --- a/src/openhuman/memory_diff/ops.rs +++ b/src/openhuman/memory_diff/ops.rs @@ -1,118 +1,78 @@ //! Business logic for memory diff: snapshot capture, diff computation, -//! checkpoints, and cleanup. +//! checkpoints, and cleanup — all backed by the git ledger (`git_store`). +//! +//! `mem_tree_chunks` remains authoritative. Each `take_snapshot` materialises a +//! source's current items as git blobs and records them as a commit; diffs are +//! git tree diffs, checkpoints are tags, read markers are refs. use std::collections::HashMap; -use sha2::{Digest, Sha256}; +use anyhow::{anyhow, bail}; use crate::openhuman::config::Config; use crate::openhuman::memory_sources::types::{MemorySourceEntry, SourceKind}; use crate::openhuman::memory_store::chunks::store as chunk_store; -use super::store; +use super::git_store::{Ledger, SnapshotMeta}; use super::types::*; -const DEFAULT_RETENTION_DAYS: u32 = 30; -const MAX_SNAPSHOTS_PER_SOURCE: u32 = 100; -const MAX_TEXT_DIFF_CHARS: usize = 2000; -/// Upper bound on per-item content persisted into a snapshot, so the "from" -/// side of a text diff survives the next sync overwriting the live chunk -/// store. Items larger than this skip content capture (`content = None`); -/// hash-based add/remove/modify detection still works for them. -const MAX_SNAPSHOT_CONTENT_BYTES: usize = 64 * 1024; - /// Take a snapshot of the current chunk-store state for a source. /// -/// Reads from `mem_tree_chunks` (already-ingested data), groups by item, -/// hashes content, and persists to the diff database. +/// Reads from `mem_tree_chunks` (already-ingested data), groups by item, and +/// commits one blob per item to the git ledger. Returns the new [`Snapshot`] +/// whose `id` is the commit SHA. pub async fn take_snapshot( source: &MemorySourceEntry, config: &Config, trigger: SnapshotTrigger, ) -> Result { - let source_clone = source.clone(); + let prefix = source_id_prefix(source); let config_clone = config.clone(); - let prefix = source_id_prefix(&source_clone); + // Group chunk content per item, in chunk order, into (item_id, content). let items = tokio::task::spawn_blocking(move || { chunk_store::with_connection(&config_clone, |conn| { let mut stmt = conn.prepare( - "SELECT source_id, content, timestamp_ms \ + "SELECT source_id, content \ FROM mem_tree_chunks \ WHERE source_id LIKE ?1 \ ORDER BY source_id, seq_in_source", )?; - let mut groups: HashMap = HashMap::new(); + let mut groups: HashMap> = HashMap::new(); let rows = stmt.query_map([&prefix], |r| { - Ok(( - r.get::<_, String>(0)?, - r.get::<_, String>(1)?, - r.get::<_, i64>(2)?, - )) + Ok((r.get::<_, String>(0)?, r.get::<_, String>(1)?)) })?; - for row in rows { - let (composite_source_id, content, ts) = row?; + let (composite_source_id, content) = row?; let item_id = extract_item_id(&composite_source_id); - let acc = groups.entry(item_id).or_default(); - acc.content_parts.push(content); - acc.max_timestamp_ms = acc.max_timestamp_ms.max(Some(ts)); - acc.chunk_count += 1; + groups.entry(item_id).or_default().push(content); } - let mut snapshot_items: Vec = groups + let mut items: Vec<(String, String)> = groups .into_iter() - .map(|(item_id, acc)| { - let concat = acc.content_parts.join(""); - let hash = sha256_hex(concat.as_bytes()); - let title = derive_title(&item_id, &concat); - // Persist bounded content so a future diff has both sides. - let content = if concat.len() <= MAX_SNAPSHOT_CONTENT_BYTES { - Some(concat) - } else { - None - }; - SnapshotItem { - item_id, - title, - content_hash: hash, - content, - timestamp_ms: acc.max_timestamp_ms, - chunk_count: acc.chunk_count, - } - }) + .map(|(item_id, parts)| (item_id, parts.join(""))) .collect(); - snapshot_items.sort_by(|a, b| a.item_id.cmp(&b.item_id)); - Ok(snapshot_items) + items.sort_by(|a, b| a.0.cmp(&b.0)); + Ok(items) }) }) .await .map_err(|e| format!("snapshot join error: {e}"))? .map_err(|e: anyhow::Error| format!("snapshot query error: {e:#}"))?; - let snapshot = Snapshot { - id: format!("snap_{}", uuid::Uuid::new_v4()), + let meta = SnapshotMeta { source_id: source.id.clone(), source_kind: source.kind.as_str().to_string(), label: source.label.clone(), trigger, - item_count: items.len() as u32, - taken_at_ms: chrono::Utc::now().timestamp_millis(), }; - let workspace_dir = config.workspace_dir.clone(); - let snap_clone = snapshot.clone(); - let items_clone = items.clone(); - tokio::task::spawn_blocking(move || { - store::with_connection(&workspace_dir, |conn| { - store::insert_snapshot(conn, &snap_clone, &items_clone)?; + let now_ms = chrono::Utc::now().timestamp_millis(); - let cutoff = chrono::Utc::now().timestamp_millis() - - (DEFAULT_RETENTION_DAYS as i64 * 24 * 60 * 60 * 1000); - store::cleanup_old_snapshots(conn, cutoff, MAX_SNAPSHOTS_PER_SOURCE)?; - Ok(()) - }) + let snapshot = tokio::task::spawn_blocking(move || -> anyhow::Result { + let ledger = Ledger::open(&workspace_dir)?; + ledger.commit_snapshot(&meta, &items, now_ms) }) .await .map_err(|e| format!("snapshot persist join: {e}"))? @@ -158,121 +118,50 @@ pub async fn compute_diff( let to_id = to_snapshot_id.to_string(); let from_id = from_snapshot_id.map(|s| s.to_string()); - let (to_snap, from_snap, to_items, from_items) = tokio::task::spawn_blocking(move || { - store::with_connection(&workspace_dir, |conn| { - let to_snap = store::get_snapshot(conn, &to_id)? - .ok_or_else(|| anyhow::anyhow!("snapshot not found: {to_id}"))?; - let to_items = store::get_snapshot_items(conn, &to_id)?; + tokio::task::spawn_blocking(move || -> anyhow::Result { + let ledger = Ledger::open(&workspace_dir)?; + let to_snap = ledger + .get_snapshot(&to_id)? + .ok_or_else(|| anyhow!("snapshot not found: {to_id}"))?; - let (from_snap, from_items) = match &from_id { - Some(fid) => { - let s = store::get_snapshot(conn, fid)? - .ok_or_else(|| anyhow::anyhow!("snapshot not found: {fid}"))?; - if s.source_id != to_snap.source_id { - anyhow::bail!( - "cross-source diff not allowed: from={} to={}", - s.source_id, - to_snap.source_id - ); - } - let items = store::get_snapshot_items(conn, fid)?; - (Some(s), items) + let from_snap = match &from_id { + Some(fid) => { + let s = ledger + .get_snapshot(fid)? + .ok_or_else(|| anyhow!("snapshot not found: {fid}"))?; + if s.source_id != to_snap.source_id { + bail!( + "cross-source diff not allowed: from={} to={}", + s.source_id, + to_snap.source_id + ); } - None => (None, Vec::new()), - }; + Some(s) + } + None => None, + }; - Ok((to_snap, from_snap, to_items, from_items)) + let (changes, summary) = ledger.compute_changes( + from_id.as_deref(), + &to_id, + &to_snap.source_id, + to_snap.item_count, + include_text_diff, + )?; + + Ok(DiffResult { + source_id: to_snap.source_id.clone(), + source_kind: to_snap.source_kind.clone(), + source_label: to_snap.label.clone(), + from_snapshot_id: from_snap.map(|s| s.id), + to_snapshot_id: to_snap.id.clone(), + summary, + changes, }) }) .await .map_err(|e| format!("diff join: {e}"))? - .map_err(|e: anyhow::Error| format!("diff load: {e:#}"))?; - - let from_map: HashMap<&str, &SnapshotItem> = - from_items.iter().map(|i| (i.item_id.as_str(), i)).collect(); - let to_map: HashMap<&str, &SnapshotItem> = - to_items.iter().map(|i| (i.item_id.as_str(), i)).collect(); - - let mut changes = Vec::new(); - let mut summary = DiffSummary::default(); - - // Added + Modified - for to_item in &to_items { - match from_map.get(to_item.item_id.as_str()) { - None => { - summary.added += 1; - changes.push(ItemChange { - item_id: to_item.item_id.clone(), - title: to_item.title.clone(), - kind: ChangeKind::Added, - old_content_hash: None, - new_content_hash: Some(to_item.content_hash.clone()), - text_diff: None, - }); - } - Some(from_item) => { - if from_item.content_hash != to_item.content_hash { - summary.modified += 1; - changes.push(ItemChange { - item_id: to_item.item_id.clone(), - title: to_item.title.clone(), - kind: ChangeKind::Modified, - old_content_hash: Some(from_item.content_hash.clone()), - new_content_hash: Some(to_item.content_hash.clone()), - text_diff: None, - }); - } else { - summary.unchanged += 1; - } - } - } - } - - // Removed - for from_item in &from_items { - if !to_map.contains_key(from_item.item_id.as_str()) { - summary.removed += 1; - changes.push(ItemChange { - item_id: from_item.item_id.clone(), - title: from_item.title.clone(), - kind: ChangeKind::Removed, - old_content_hash: Some(from_item.content_hash.clone()), - new_content_hash: None, - text_diff: None, - }); - } - } - - // Compute text diffs for modified items if requested - if include_text_diff { - let modified_ids: Vec = changes - .iter() - .filter(|c| c.kind == ChangeKind::Modified) - .map(|c| c.item_id.clone()) - .collect(); - - if !modified_ids.is_empty() { - let text_diffs = compute_text_diffs_from_snapshots(&from_map, &to_map, &modified_ids); - - for change in &mut changes { - if change.kind == ChangeKind::Modified { - if let Some(diff_text) = text_diffs.get(&change.item_id) { - change.text_diff = Some(truncate(diff_text, MAX_TEXT_DIFF_CHARS)); - } - } - } - } - } - - Ok(DiffResult { - source_id: to_snap.source_id.clone(), - source_kind: to_snap.source_kind.clone(), - source_label: to_snap.label.clone(), - from_snapshot_id: from_snap.map(|s| s.id), - to_snapshot_id: to_snap.id.clone(), - summary, - changes, - }) + .map_err(|e: anyhow::Error| format!("compute_diff: {e:#}")) } /// Diff current state (latest snapshot) vs previous snapshot for a source. @@ -284,10 +173,9 @@ pub async fn diff_since_last( let workspace_dir = config.workspace_dir.clone(); let source_id = source.id.clone(); - let snapshots = tokio::task::spawn_blocking(move || { - store::with_connection(&workspace_dir, |conn| { - store::latest_snapshots_for_source(conn, &source_id, 2) - }) + let snapshots = tokio::task::spawn_blocking(move || -> anyhow::Result> { + let ledger = Ledger::open(&workspace_dir)?; + ledger.latest_snapshots_for_source(&source_id, 2) }) .await .map_err(|e| format!("diff_since_last join: {e}"))? @@ -311,8 +199,8 @@ pub async fn diff_since_last( /// Diff a source's latest snapshot against its read marker — i.e. everything /// that changed since the agent last *read* this source's diff. /// -/// When `commit` is true, the read marker is advanced to the head snapshot -/// after the diff is computed, so a subsequent call returns only newer +/// When `commit` is true, the read marker (a git ref) is advanced to the head +/// snapshot after the diff is computed, so a subsequent call returns only newer /// changes. This is the turn-to-turn primitive: read the world delta, then /// acknowledge it as consumed. pub async fn diff_since_read( @@ -325,45 +213,37 @@ pub async fn diff_since_read( let source_id = source.id.clone(); // Resolve head (latest snapshot) and the marker's base snapshot. If the - // marker points at a snapshot that has since been cleaned up, treat it as - // unread (base = None) rather than erroring. - let (head, base_id) = tokio::task::spawn_blocking(move || { - store::with_connection(&workspace_dir, |conn| { - let head = store::latest_snapshots_for_source(conn, &source_id, 1)? + // marker points at a commit that no longer resolves, treat it as unread. + let (head, base_id) = tokio::task::spawn_blocking( + move || -> anyhow::Result<(Option, Option)> { + let ledger = Ledger::open(&workspace_dir)?; + let head = ledger + .latest_snapshots_for_source(&source_id, 1)? .into_iter() .next(); - let marker = store::get_read_marker(conn, &source_id)?; + let marker = ledger.get_read_marker(&source_id)?; let base_id = match marker { - Some(snap_id) if store::get_snapshot(conn, &snap_id)?.is_some() => Some(snap_id), + Some(snap_id) if ledger.get_snapshot(&snap_id)?.is_some() => Some(snap_id), _ => None, }; Ok((head, base_id)) - }) - }) + }, + ) .await .map_err(|e| format!("diff_since_read join: {e}"))? .map_err(|e: anyhow::Error| format!("diff_since_read: {e:#}"))?; let head = head.ok_or_else(|| "no snapshots found for this source".to_string())?; - // Marker already at head → nothing new since last read. - let from_id = match &base_id { - Some(id) if *id == head.id => Some(head.id.as_str()), - Some(id) => Some(id.as_str()), - None => None, - }; - - let diff = compute_diff(config, from_id, &head.id, include_text_diff).await?; + let diff = compute_diff(config, base_id.as_deref(), &head.id, include_text_diff).await?; if commit { let workspace_dir = config.workspace_dir.clone(); let source_id = source.id.clone(); let head_id = head.id.clone(); - let now_ms = chrono::Utc::now().timestamp_millis(); - tokio::task::spawn_blocking(move || { - store::with_connection(&workspace_dir, |conn| { - store::upsert_read_marker(conn, &source_id, &head_id, now_ms) - }) + tokio::task::spawn_blocking(move || -> anyhow::Result<()> { + let ledger = Ledger::open(&workspace_dir)?; + ledger.set_read_marker(&source_id, &head_id) }) .await .map_err(|e| format!("diff_since_read commit join: {e}"))? @@ -399,27 +279,27 @@ pub async fn mark_read(config: &Config, source_ids: Option>) -> Resu let workspace_dir = config.workspace_dir.clone(); let ids_for_blocking = target_ids.clone(); - let (marked, snapshot_ids) = tokio::task::spawn_blocking(move || { - store::with_connection(&workspace_dir, |conn| { - let now_ms = chrono::Utc::now().timestamp_millis(); + let (marked, snapshot_ids) = + tokio::task::spawn_blocking(move || -> anyhow::Result<(u64, Vec)> { + let ledger = Ledger::open(&workspace_dir)?; let mut count = 0u64; let mut snapshot_ids = Vec::new(); for sid in &ids_for_blocking { - if let Some(head) = store::latest_snapshots_for_source(conn, sid, 1)? + if let Some(head) = ledger + .latest_snapshots_for_source(sid, 1)? .into_iter() .next() { - store::upsert_read_marker(conn, sid, &head.id, now_ms)?; + ledger.set_read_marker(sid, &head.id)?; snapshot_ids.push(head.id); count += 1; } } Ok((count, snapshot_ids)) }) - }) - .await - .map_err(|e| format!("mark_read join: {e}"))? - .map_err(|e: anyhow::Error| format!("mark_read: {e:#}"))?; + .await + .map_err(|e| format!("mark_read join: {e}"))? + .map_err(|e: anyhow::Error| format!("mark_read: {e:#}"))?; tracing::debug!( sources = marked, @@ -436,66 +316,62 @@ pub async fn mark_read(config: &Config, source_ids: Option>) -> Resu Ok(marked) } -/// Create a checkpoint that groups the latest snapshot per enabled source. +/// Create a checkpoint (git tag at HEAD) grouping the latest snapshot per +/// enabled source. pub async fn create_checkpoint(label: &str, config: &Config) -> Result { let sources = crate::openhuman::memory_sources::registry::list_sources() .await .map_err(|e| format!("list sources: {e}"))?; - let enabled: Vec<_> = sources.into_iter().filter(|s| s.enabled).collect(); - // Take snapshots for any source that doesn't have one yet - for source in &enabled { - let workspace_dir = config.workspace_dir.clone(); - let sid = source.id.clone(); - let has_snapshot = tokio::task::spawn_blocking(move || { - store::with_connection(&workspace_dir, |conn| { - let snaps = store::latest_snapshots_for_source(conn, &sid, 1)?; - Ok(!snaps.is_empty()) - }) - }) - .await - .map_err(|e| format!("checkpoint check join: {e}"))? - .map_err(|e: anyhow::Error| format!("checkpoint check: {e:#}"))?; - - if !has_snapshot { - take_snapshot(source, config, SnapshotTrigger::Manual).await?; - } - } - - // Collect latest snapshot ID per source + // Take a snapshot for any source that doesn't have one yet, so the + // checkpoint has a baseline for every source. let workspace_dir = config.workspace_dir.clone(); - let source_ids: Vec = enabled.iter().map(|s| s.id.clone()).collect(); - let snapshot_ids = tokio::task::spawn_blocking(move || { - store::with_connection(&workspace_dir, |conn| { - let mut ids = Vec::new(); - for sid in &source_ids { - if let Some(snap) = store::latest_snapshots_for_source(conn, sid, 1)? - .into_iter() - .next() - { - ids.push(snap.id); - } + let enabled_ids: Vec = enabled.iter().map(|s| s.id.clone()).collect(); + let ids_clone = enabled_ids.clone(); + let lacking = tokio::task::spawn_blocking(move || -> anyhow::Result> { + let ledger = Ledger::open(&workspace_dir)?; + let mut lacking = Vec::new(); + for sid in &ids_clone { + if ledger.snapshot_count_for_source(sid)? == 0 { + lacking.push(sid.clone()); } - Ok(ids) - }) + } + Ok(lacking) }) .await - .map_err(|e| format!("checkpoint gather join: {e}"))? - .map_err(|e: anyhow::Error| format!("checkpoint gather: {e:#}"))?; + .map_err(|e| format!("checkpoint check join: {e}"))? + .map_err(|e: anyhow::Error| format!("checkpoint check: {e:#}"))?; - let checkpoint = Checkpoint { - id: format!("ckpt_{}", uuid::Uuid::new_v4()), - label: label.to_string(), - created_at_ms: chrono::Utc::now().timestamp_millis(), - snapshot_ids: snapshot_ids.clone(), - }; + for source in enabled.iter().filter(|s| lacking.contains(&s.id)) { + take_snapshot(source, config, SnapshotTrigger::Manual).await?; + } + // Gather the latest snapshot id per source, then tag HEAD. let workspace_dir = config.workspace_dir.clone(); - let ckpt_clone = checkpoint.clone(); - tokio::task::spawn_blocking(move || { - store::with_connection(&workspace_dir, |conn| { - store::insert_checkpoint(conn, &ckpt_clone) + let checkpoint_id = format!("ckpt_{}", uuid::Uuid::new_v4()); + let created_at_ms = chrono::Utc::now().timestamp_millis(); + let label_owned = label.to_string(); + let ckpt_id_clone = checkpoint_id.clone(); + + let checkpoint = tokio::task::spawn_blocking(move || -> anyhow::Result { + let ledger = Ledger::open(&workspace_dir)?; + let mut snapshot_ids = Vec::new(); + for sid in &enabled_ids { + if let Some(snap) = ledger + .latest_snapshots_for_source(sid, 1)? + .into_iter() + .next() + { + snapshot_ids.push(snap.id); + } + } + ledger.create_checkpoint(&ckpt_id_clone, &label_owned, &snapshot_ids, created_at_ms)?; + Ok(Checkpoint { + id: ckpt_id_clone, + label: label_owned, + created_at_ms, + snapshot_ids, }) }) .await @@ -519,80 +395,79 @@ pub async fn diff_since_checkpoint( ) -> Result { let workspace_dir = config.workspace_dir.clone(); let ckpt_id = checkpoint_id.to_string(); - let checkpoint = tokio::task::spawn_blocking(move || { - store::with_connection(&workspace_dir, |conn| { - store::get_checkpoint(conn, &ckpt_id)? - .ok_or_else(|| anyhow::anyhow!("checkpoint not found: {ckpt_id}")) + let computed_at_ms = chrono::Utc::now().timestamp_millis(); + + tokio::task::spawn_blocking(move || -> anyhow::Result { + let ledger = Ledger::open(&workspace_dir)?; + let checkpoint = ledger + .get_checkpoint(&ckpt_id)? + .ok_or_else(|| anyhow!("checkpoint not found: {ckpt_id}"))?; + + let mut per_source = Vec::new(); + let mut agg = DiffSummary::default(); + + for snap_id in &checkpoint.snapshot_ids { + let Some(base) = ledger.get_snapshot(snap_id)? else { + continue; + }; + let Some(head) = ledger + .latest_snapshots_for_source(&base.source_id, 1)? + .into_iter() + .next() + else { + continue; + }; + if head.id == base.id { + continue; // unchanged since the checkpoint + } + + let (changes, summary) = ledger.compute_changes( + Some(&base.id), + &head.id, + &head.source_id, + head.item_count, + include_text_diff, + )?; + agg.added += summary.added; + agg.removed += summary.removed; + agg.modified += summary.modified; + agg.unchanged += summary.unchanged; + per_source.push(DiffResult { + source_id: head.source_id.clone(), + source_kind: head.source_kind.clone(), + source_label: head.label.clone(), + from_snapshot_id: Some(base.id.clone()), + to_snapshot_id: head.id.clone(), + summary, + changes, + }); + } + + Ok(CrossSourceDiff { + checkpoint_id: Some(checkpoint.id), + computed_at_ms, + summary: agg, + per_source, }) }) .await - .map_err(|e| format!("checkpoint load join: {e}"))? - .map_err(|e: anyhow::Error| format!("checkpoint load: {e:#}"))?; - - // For each snapshot in the checkpoint, find the source's latest snapshot - let workspace_dir = config.workspace_dir.clone(); - let snap_ids = checkpoint.snapshot_ids.clone(); - let snapshot_pairs: Vec<(Snapshot, Option)> = - tokio::task::spawn_blocking(move || { - store::with_connection(&workspace_dir, |conn| { - let mut pairs = Vec::new(); - for snap_id in &snap_ids { - let base_snap = store::get_snapshot(conn, snap_id)?; - if let Some(base) = base_snap { - let latest = store::latest_snapshots_for_source(conn, &base.source_id, 1)? - .into_iter() - .next(); - if let Some(head) = latest { - if head.id != base.id { - pairs.push((head, Some(base))); - } - // Same snapshot = no changes, skip - } - } - } - Ok(pairs) - }) - }) - .await - .map_err(|e| format!("checkpoint pairs join: {e}"))? - .map_err(|e: anyhow::Error| format!("checkpoint pairs: {e:#}"))?; - - let mut per_source = Vec::new(); - let mut agg = DiffSummary::default(); - - for (head, base) in &snapshot_pairs { - let diff = compute_diff( - config, - base.as_ref().map(|s| s.id.as_str()), - &head.id, - include_text_diff, - ) - .await?; - agg.added += diff.summary.added; - agg.removed += diff.summary.removed; - agg.modified += diff.summary.modified; - agg.unchanged += diff.summary.unchanged; - per_source.push(diff); - } - - Ok(CrossSourceDiff { - checkpoint_id: Some(checkpoint.id), - computed_at_ms: chrono::Utc::now().timestamp_millis(), - summary: agg, - per_source, - }) + .map_err(|e| format!("diff_since_checkpoint join: {e}"))? + .map_err(|e: anyhow::Error| format!("diff_since_checkpoint: {e:#}")) } -/// Delete snapshots older than `days` days. +/// Delete checkpoint tags older than `older_than_days`. +/// +/// Snapshot commits are retained — git history *is* the ledger, and git's +/// delta compression keeps it compact — so cleanup only prunes named baselines. +/// Returns the number of checkpoints deleted. pub async fn cleanup(config: &Config, older_than_days: u32) -> Result { let workspace_dir = config.workspace_dir.clone(); let cutoff = chrono::Utc::now().timestamp_millis() - (older_than_days as i64 * 24 * 60 * 60 * 1000); - tokio::task::spawn_blocking(move || { - store::with_connection(&workspace_dir, |conn| { - store::cleanup_old_snapshots(conn, cutoff, MAX_SNAPSHOTS_PER_SOURCE) - }) + tokio::task::spawn_blocking(move || -> anyhow::Result { + let ledger = Ledger::open(&workspace_dir)?; + ledger.cleanup_checkpoints(cutoff) }) .await .map_err(|e| format!("cleanup join: {e}"))? @@ -632,86 +507,10 @@ fn extract_item_id(composite: &str) -> String { composite.to_string() } -fn sha256_hex(bytes: &[u8]) -> String { - let mut hasher = Sha256::new(); - hasher.update(bytes); - hex::encode(hasher.finalize()) -} - -fn truncate(s: &str, max_chars: usize) -> String { - if s.len() <= max_chars { - s.to_string() - } else { - let mut end = max_chars; - while !s.is_char_boundary(end) && end > 0 { - end -= 1; - } - format!("{}…(truncated)", &s[..end]) - } -} - -/// Derive a human-readable title for an item from its content. -/// -/// Uses the first non-empty line (a Markdown heading marker is stripped), -/// trimmed and bounded. Falls back to the item id when no usable line exists -/// (e.g. binary or empty content) so the diff output is never blank. -fn derive_title(item_id: &str, content: &str) -> String { - let first_line = content - .lines() - .map(str::trim) - .find(|l| !l.is_empty()) - .map(|l| l.trim_start_matches('#').trim()); - - match first_line { - Some(l) if !l.is_empty() => truncate(l, 120), - _ => item_id.to_string(), - } -} - -/// Compute unified text diffs for modified items from the content stored in -/// the two snapshots being compared. Both sides are read from the diff DB -/// (bounded content captured at snapshot time), so this works even after the -/// live chunk store has been overwritten by a later sync. Items whose content -/// was too large to capture (`content = None` on either side) are skipped. -fn compute_text_diffs_from_snapshots( - from_items: &HashMap<&str, &SnapshotItem>, - to_items: &HashMap<&str, &SnapshotItem>, - item_ids: &[String], -) -> HashMap { - let mut out = HashMap::new(); - for item_id in item_ids { - let (Some(from), Some(to)) = ( - from_items.get(item_id.as_str()), - to_items.get(item_id.as_str()), - ) else { - continue; - }; - let (Some(old), Some(new)) = (from.content.as_deref(), to.content.as_deref()) else { - continue; - }; - let diff = similar::TextDiff::from_lines(old, new); - let unified = diff - .unified_diff() - .context_radius(3) - .header("before", "after") - .to_string(); - if !unified.trim().is_empty() { - out.insert(item_id.clone(), unified); - } - } - out -} - -#[derive(Default)] -struct ItemAccumulator { - content_parts: Vec, - max_timestamp_ms: Option, - chunk_count: u32, -} - #[cfg(test)] mod tests { use super::*; + use crate::openhuman::memory_diff::git_store::Ledger; #[test] fn extract_item_id_reader_backed() { @@ -737,97 +536,21 @@ mod tests { #[test] fn source_id_prefix_folder() { - let entry = MemorySourceEntry { - id: "src_abc".into(), - kind: SourceKind::Folder, - label: "x".into(), - enabled: true, - toolkit: None, - connection_id: None, - path: Some("/tmp".into()), - glob: None, - url: None, - branch: None, - paths: Vec::new(), - query: None, - since_days: None, - max_items: None, - max_commits: None, - max_issues: None, - max_prs: None, - selector: None, - max_tokens_per_sync: None, - max_cost_per_sync_usd: None, - sync_depth_days: None, - }; - assert_eq!(source_id_prefix(&entry), "mem_src:src_abc:%"); - } - - #[test] - fn source_id_prefix_composio() { - let entry = MemorySourceEntry { - id: "src_cmp".into(), - kind: SourceKind::Composio, - label: "Gmail".into(), - enabled: true, - toolkit: Some("gmail".into()), - connection_id: Some("cmp_1".into()), - path: None, - glob: None, - url: None, - branch: None, - paths: Vec::new(), - query: None, - since_days: None, - max_items: None, - max_commits: None, - max_issues: None, - max_prs: None, - selector: None, - max_tokens_per_sync: None, - max_cost_per_sync_usd: None, - sync_depth_days: None, - }; - assert_eq!(source_id_prefix(&entry), "gmail:%"); - } - - #[test] - fn truncate_short_string() { - assert_eq!(truncate("hello", 10), "hello"); - } - - #[test] - fn truncate_long_string() { - let s = "a".repeat(100); - let t = truncate(&s, 50); - assert!(t.len() < 70); - assert!(t.ends_with("…(truncated)")); - } - - #[test] - fn sha256_hex_deterministic() { - let h1 = sha256_hex(b"hello world"); - let h2 = sha256_hex(b"hello world"); - assert_eq!(h1, h2); - assert_ne!(sha256_hex(b"hello"), sha256_hex(b"world")); - } - - #[test] - fn derive_title_uses_first_nonempty_line() { - assert_eq!(derive_title("file.md", "# Heading\nbody"), "Heading"); assert_eq!( - derive_title("file.md", "\n\n Plain title \nmore"), - "Plain title" + source_id_prefix(&folder_source("src_abc")), + "mem_src:src_abc:%" ); } #[test] - fn derive_title_falls_back_to_item_id() { - assert_eq!(derive_title("doc_42", ""), "doc_42"); - assert_eq!(derive_title("doc_42", " \n "), "doc_42"); + fn source_id_prefix_composio() { + let mut entry = folder_source("src_cmp"); + entry.kind = SourceKind::Composio; + entry.toolkit = Some("gmail".into()); + assert_eq!(source_id_prefix(&entry), "gmail:%"); } - // ── Integration-style ops tests over a temp diff.db ─────────────────── + // ── Integration-style ops tests over a temp git ledger ──────────────── fn test_config() -> Config { let dir = tempfile::tempdir().unwrap(); @@ -838,33 +561,6 @@ mod tests { config } - fn item(item_id: &str, hash: &str, content: &str) -> SnapshotItem { - SnapshotItem { - item_id: item_id.to_string(), - title: item_id.to_string(), - content_hash: hash.to_string(), - content: Some(content.to_string()), - timestamp_ms: Some(1000), - chunk_count: 1, - } - } - - fn seed(config: &Config, id: &str, source_id: &str, taken_at_ms: i64, items: &[SnapshotItem]) { - let snap = Snapshot { - id: id.to_string(), - source_id: source_id.to_string(), - source_kind: "folder".to_string(), - label: "Docs".to_string(), - trigger: SnapshotTrigger::Auto, - item_count: items.len() as u32, - taken_at_ms, - }; - store::with_connection(&config.workspace_dir, |conn| { - store::insert_snapshot(conn, &snap, items) - }) - .unwrap(); - } - fn folder_source(id: &str) -> MemorySourceEntry { MemorySourceEntry { id: id.into(), @@ -891,35 +587,49 @@ mod tests { } } + /// Seed a snapshot directly through the ledger (bypassing the chunk store). + fn seed( + config: &Config, + source_id: &str, + taken_at_ms: i64, + items: &[(&str, &str)], + ) -> Snapshot { + let ledger = Ledger::open(&config.workspace_dir).unwrap(); + let items: Vec<(String, String)> = items + .iter() + .map(|(k, v)| (k.to_string(), v.to_string())) + .collect(); + ledger + .commit_snapshot( + &SnapshotMeta { + source_id: source_id.to_string(), + source_kind: "folder".to_string(), + label: "Docs".to_string(), + trigger: SnapshotTrigger::Auto, + }, + &items, + taken_at_ms, + ) + .unwrap() + } + #[tokio::test] async fn compute_diff_detects_added_modified_removed() { let config = test_config(); - // from: a(h1), b(h2), c(h3) - seed( + let from = seed( &config, - "snap_from", "src_a", 1000, - &[ - item("a", "h1", "alpha"), - item("b", "h2", "beta"), - item("c", "h3", "gamma"), - ], + &[("a", "alpha"), ("b", "beta"), ("c", "gamma")], ); - // to: a(h1, unchanged), b(h2b, modified), c removed, d(h4, added) - seed( + let to = seed( &config, - "snap_to", "src_a", 2000, - &[ - item("a", "h1", "alpha"), - item("b", "h2b", "beta v2"), - item("d", "h4", "delta"), - ], + &[("a", "alpha"), ("b", "beta v2"), ("d", "delta")], ); - let diff = compute_diff(&config, Some("snap_from"), "snap_to", false) + let diff = compute_diff(&config, Some(&from.id), &to.id, false) .await .unwrap(); @@ -943,8 +653,8 @@ mod tests { #[tokio::test] async fn compute_diff_against_none_marks_all_added() { let config = test_config(); - seed(&config, "snap_to", "src_a", 1000, &[item("a", "h1", "x")]); - let diff = compute_diff(&config, None, "snap_to", false).await.unwrap(); + let to = seed(&config, "src_a", 1000, &[("a", "x")]); + let diff = compute_diff(&config, None, &to.id, false).await.unwrap(); assert_eq!(diff.summary.added, 1); assert_eq!(diff.from_snapshot_id, None); } @@ -952,9 +662,9 @@ mod tests { #[tokio::test] async fn compute_diff_rejects_cross_source() { let config = test_config(); - seed(&config, "from_a", "src_a", 1000, &[]); - seed(&config, "to_b", "src_b", 2000, &[]); - let err = compute_diff(&config, Some("from_a"), "to_b", false) + let from = seed(&config, "src_a", 1000, &[("a", "x")]); + let to = seed(&config, "src_b", 2000, &[("b", "y")]); + let err = compute_diff(&config, Some(&from.id), &to.id, false) .await .unwrap_err(); assert!(err.contains("cross-source"), "got: {err}"); @@ -963,25 +673,22 @@ mod tests { #[tokio::test] async fn compute_diff_text_diff_only_when_requested() { let config = test_config(); - seed( + let from = seed(&config, "src_a", 1000, &[("a", "line one\nline two\n")]); + let to = seed( &config, - "f", - "src_a", - 1000, - &[item("a", "h1", "line one\nline two\n")], - ); - seed( - &config, - "t", "src_a", 2000, - &[item("a", "h2", "line one\nline TWO changed\n")], + &[("a", "line one\nline TWO changed\n")], ); - let without = compute_diff(&config, Some("f"), "t", false).await.unwrap(); + let without = compute_diff(&config, Some(&from.id), &to.id, false) + .await + .unwrap(); assert!(without.changes[0].text_diff.is_none()); - let with = compute_diff(&config, Some("f"), "t", true).await.unwrap(); + let with = compute_diff(&config, Some(&from.id), &to.id, true) + .await + .unwrap(); let td = with.changes[0] .text_diff .as_ref() @@ -998,18 +705,12 @@ mod tests { assert!(diff_since_last(&source, &config, false).await.is_err()); // 1 snapshot → everything added (diff vs None) - seed(&config, "s1", "src_a", 1000, &[item("a", "h1", "x")]); + seed(&config, "src_a", 1000, &[("a", "x")]); let one = diff_since_last(&source, &config, false).await.unwrap(); assert_eq!(one.summary.added, 1); // 2 snapshots → diff latest vs previous - seed( - &config, - "s2", - "src_a", - 2000, - &[item("a", "h1", "x"), item("b", "h2", "y")], - ); + seed(&config, "src_a", 2000, &[("a", "x"), ("b", "y")]); let two = diff_since_last(&source, &config, false).await.unwrap(); assert_eq!(two.summary.added, 1, "b is new in s2"); assert_eq!(two.summary.unchanged, 1, "a unchanged"); @@ -1020,7 +721,7 @@ mod tests { let config = test_config(); let source = folder_source("src_a"); - seed(&config, "s1", "src_a", 1000, &[item("a", "h1", "x")]); + seed(&config, "src_a", 1000, &[("a", "x")]); // First read: no marker → full diff (a added), and commit advances marker. let first = diff_since_read(&source, &config, false, true) @@ -1038,13 +739,7 @@ mod tests { assert!(second.changes.is_empty()); // New snapshot then read: only the delta since the marker shows. - seed( - &config, - "s2", - "src_a", - 2000, - &[item("a", "h1", "x"), item("b", "h2", "y")], - ); + seed(&config, "src_a", 2000, &[("a", "x"), ("b", "y")]); let third = diff_since_read(&source, &config, false, true) .await .unwrap(); @@ -1056,7 +751,7 @@ mod tests { async fn diff_since_read_without_commit_does_not_advance_marker() { let config = test_config(); let source = folder_source("src_a"); - seed(&config, "s1", "src_a", 1000, &[item("a", "h1", "x")]); + seed(&config, "src_a", 1000, &[("a", "x")]); // Preview (commit=false) twice → both show the full diff. let a = diff_since_read(&source, &config, false, false) @@ -1073,7 +768,7 @@ mod tests { async fn mark_read_advances_marker_for_explicit_sources() { let config = test_config(); let source = folder_source("src_a"); - seed(&config, "s1", "src_a", 1000, &[item("a", "h1", "x")]); + seed(&config, "src_a", 1000, &[("a", "x")]); let marked = mark_read(&config, Some(vec!["src_a".to_string()])) .await @@ -1092,21 +787,17 @@ mod tests { async fn diff_since_checkpoint_aggregates_across_sources() { let config = test_config(); // Baseline snapshots for two sources, grouped into a checkpoint. - seed(&config, "a1", "src_a", 1000, &[item("a", "h1", "x")]); - seed(&config, "b1", "src_b", 1000, &[item("b", "h1", "y")]); - let ckpt = Checkpoint { - id: "ckpt_1".to_string(), - label: "base".to_string(), - created_at_ms: 1500, - snapshot_ids: vec!["a1".to_string(), "b1".to_string()], - }; - store::with_connection(&config.workspace_dir, |conn| { - store::insert_checkpoint(conn, &ckpt) - }) - .unwrap(); + let a1 = seed(&config, "src_a", 1000, &[("a", "x")]); + let b1 = seed(&config, "src_b", 1000, &[("b", "y")]); + { + let ledger = Ledger::open(&config.workspace_dir).unwrap(); + ledger + .create_checkpoint("ckpt_1", "base", &[a1.id.clone(), b1.id.clone()], 1500) + .unwrap(); + } - // src_a gets a new head with a modification; src_b unchanged (no new head). - seed(&config, "a2", "src_a", 2000, &[item("a", "h2", "x v2")]); + // src_a gets a new head with a modification; src_b unchanged. + seed(&config, "src_a", 2000, &[("a", "x v2")]); let cross = diff_since_checkpoint("ckpt_1", &config, false) .await diff --git a/src/openhuman/memory_diff/rpc.rs b/src/openhuman/memory_diff/rpc.rs index 2aa6c363d..ae3d506e4 100644 --- a/src/openhuman/memory_diff/rpc.rs +++ b/src/openhuman/memory_diff/rpc.rs @@ -6,8 +6,8 @@ use serde::{Deserialize, Serialize}; use crate::openhuman::config::rpc as config_rpc; use crate::rpc::RpcOutcome; +use super::git_store::Ledger; use super::ops; -use super::store; use super::types::*; // ── Request / Response types ────────────────────────────────────────── @@ -166,10 +166,9 @@ pub async fn list_snapshots_rpc( let limit = req.limit.unwrap_or(50) as u32; let source_id = req.source_id; - let snapshots = tokio::task::spawn_blocking(move || { - store::with_connection(&workspace_dir, |conn| { - store::list_snapshots(conn, source_id.as_deref(), limit) - }) + let snapshots = tokio::task::spawn_blocking(move || -> anyhow::Result> { + let ledger = Ledger::open(&workspace_dir)?; + ledger.list_snapshots(source_id.as_deref(), limit) }) .await .map_err(|e| format!("list_snapshots join: {e}"))? @@ -286,8 +285,9 @@ pub async fn list_checkpoints_rpc( let workspace_dir = config.workspace_dir.clone(); let limit = req.limit.unwrap_or(20) as u32; - let checkpoints = tokio::task::spawn_blocking(move || { - store::with_connection(&workspace_dir, |conn| store::list_checkpoints(conn, limit)) + let checkpoints = tokio::task::spawn_blocking(move || -> anyhow::Result> { + let ledger = Ledger::open(&workspace_dir)?; + ledger.list_checkpoints(limit) }) .await .map_err(|e| format!("list_checkpoints join: {e}"))? diff --git a/src/openhuman/memory_diff/store.rs b/src/openhuman/memory_diff/store.rs deleted file mode 100644 index b67f0a83d..000000000 --- a/src/openhuman/memory_diff/store.rs +++ /dev/null @@ -1,613 +0,0 @@ -//! SQLite persistence for memory diff snapshots and checkpoints. -//! -//! Own database at `/memory_diff/diff.db`. Follows the -//! `subconscious/store.rs` self-contained pattern: opens the database, -//! runs DDL on every connection, and provides pure functions. - -use anyhow::{Context, Result}; -use rusqlite::{params, Connection, OptionalExtension}; -use std::path::Path; -use std::time::Duration; - -use super::types::{Checkpoint, Snapshot, SnapshotItem, SnapshotTrigger}; - -const BUSY_TIMEOUT: Duration = Duration::from_millis(5000); -const OPEN_RETRY_ATTEMPTS: u32 = 3; -const OPEN_RETRY_BASE_MS: u64 = 100; - -const SCHEMA_DDL: &str = " -CREATE TABLE IF NOT EXISTS mem_diff_snapshots ( - id TEXT PRIMARY KEY, - source_id TEXT NOT NULL, - source_kind TEXT NOT NULL, - label TEXT NOT NULL, - trigger_kind TEXT NOT NULL DEFAULT 'auto', - item_count INTEGER NOT NULL, - taken_at_ms INTEGER NOT NULL, - created_at_ms INTEGER NOT NULL -); -CREATE INDEX IF NOT EXISTS idx_snap_source - ON mem_diff_snapshots(source_id, taken_at_ms); - -CREATE TABLE IF NOT EXISTS mem_diff_snapshot_items ( - snapshot_id TEXT NOT NULL REFERENCES mem_diff_snapshots(id) ON DELETE CASCADE, - item_id TEXT NOT NULL, - title TEXT NOT NULL DEFAULT '', - content_hash TEXT NOT NULL, - content TEXT, - timestamp_ms INTEGER, - chunk_count INTEGER NOT NULL DEFAULT 1, - PRIMARY KEY (snapshot_id, item_id) -); - -CREATE TABLE IF NOT EXISTS mem_diff_checkpoints ( - id TEXT PRIMARY KEY, - label TEXT NOT NULL, - created_at_ms INTEGER NOT NULL -); - -CREATE TABLE IF NOT EXISTS mem_diff_checkpoint_snapshots ( - checkpoint_id TEXT NOT NULL REFERENCES mem_diff_checkpoints(id) ON DELETE CASCADE, - snapshot_id TEXT NOT NULL REFERENCES mem_diff_snapshots(id) ON DELETE CASCADE, - PRIMARY KEY (checkpoint_id, snapshot_id) -); - -CREATE TABLE IF NOT EXISTS mem_diff_read_markers ( - source_id TEXT PRIMARY KEY, - snapshot_id TEXT NOT NULL, - marked_at_ms INTEGER NOT NULL -); -"; - -pub fn with_connection( - workspace_dir: &Path, - f: impl FnOnce(&Connection) -> Result, -) -> Result { - let db_path = workspace_dir.join("memory_diff").join("diff.db"); - if let Some(parent) = db_path.parent() { - std::fs::create_dir_all(parent) - .with_context(|| format!("failed to create memory_diff dir: {}", parent.display()))?; - } - let conn = open_with_retry(&db_path)?; - f(&conn) -} - -fn open_with_retry(db_path: &Path) -> Result { - let mut last_err: Option = None; - - for attempt in 0..=OPEN_RETRY_ATTEMPTS { - match open_and_init(db_path) { - Ok(conn) => return Ok(conn), - Err(e) => { - if !is_sqlite_busy(&e) || attempt == OPEN_RETRY_ATTEMPTS { - last_err = Some(e); - break; - } - let sleep_ms = OPEN_RETRY_BASE_MS - .saturating_mul(3u64.saturating_pow(attempt)) - .min(900); - tracing::warn!( - attempt = attempt + 1, - sleep_ms = sleep_ms, - "[memory_diff::store] SQLite busy on open; retrying" - ); - std::thread::sleep(Duration::from_millis(sleep_ms)); - last_err = Some(e); - } - } - } - Err(last_err.expect("at least one attempt")) -} - -fn open_and_init(db_path: &Path) -> Result { - let conn = Connection::open(db_path) - .with_context(|| format!("failed to open memory_diff DB: {}", db_path.display()))?; - conn.busy_timeout(BUSY_TIMEOUT) - .context("configure memory_diff busy_timeout")?; - apply_journal_mode(&conn); - conn.execute_batch("PRAGMA foreign_keys = ON;") - .context("enable foreign keys")?; - conn.execute_batch(SCHEMA_DDL) - .context("failed to run memory_diff schema DDL")?; - apply_column_migrations(&conn)?; - Ok(conn) -} - -/// Idempotently add columns introduced after the initial schema. `CREATE TABLE -/// IF NOT EXISTS` is a no-op on a pre-existing `diff.db`, so newly-added -/// columns (e.g. `content` on `mem_diff_snapshot_items`) must be backfilled via -/// `ALTER TABLE ... ADD COLUMN` for workspaces created before that column existed. -fn apply_column_migrations(conn: &Connection) -> Result<()> { - if !column_exists(conn, "mem_diff_snapshot_items", "content")? { - conn.execute_batch("ALTER TABLE mem_diff_snapshot_items ADD COLUMN content TEXT;") - .context("add mem_diff_snapshot_items.content column")?; - } - Ok(()) -} - -fn column_exists(conn: &Connection, table: &str, column: &str) -> Result { - let mut stmt = conn.prepare(&format!("PRAGMA table_info({table})"))?; - let mut rows = stmt.query([])?; - while let Some(row) = rows.next()? { - // PRAGMA table_info columns: cid, name, type, notnull, dflt_value, pk. - let name: String = row.get(1)?; - if name == column { - return Ok(true); - } - } - Ok(false) -} - -fn apply_journal_mode(conn: &Connection) { - match conn.pragma_update_and_check(None, "journal_mode", "WAL", |r| r.get::<_, String>(0)) { - Ok(mode) if mode.eq_ignore_ascii_case("wal") => {} - Ok(_mode) => { - let _ = conn.pragma_update_and_check(None, "journal_mode", "TRUNCATE", |r| { - r.get::<_, String>(0) - }); - } - Err(_) => { - let _ = conn.pragma_update_and_check(None, "journal_mode", "TRUNCATE", |r| { - r.get::<_, String>(0) - }); - } - } -} - -fn is_sqlite_busy(e: &anyhow::Error) -> bool { - if let Some(sqlite_err) = e.downcast_ref::() { - matches!( - sqlite_err, - rusqlite::Error::SqliteFailure( - rusqlite::ffi::Error { - code: rusqlite::ffi::ErrorCode::DatabaseBusy - | rusqlite::ffi::ErrorCode::DatabaseLocked, - .. - }, - _ - ) - ) - } else { - false - } -} - -// ── Snapshot CRUD ────────────────────────────────────────────────────── - -pub fn insert_snapshot( - conn: &Connection, - snapshot: &Snapshot, - items: &[SnapshotItem], -) -> Result<()> { - let tx = conn.unchecked_transaction()?; - let now_ms = chrono::Utc::now().timestamp_millis(); - tx.execute( - "INSERT INTO mem_diff_snapshots (id, source_id, source_kind, label, trigger_kind, item_count, taken_at_ms, created_at_ms) - VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)", - params![ - snapshot.id, - snapshot.source_id, - snapshot.source_kind, - snapshot.label, - snapshot.trigger.as_str(), - snapshot.item_count, - snapshot.taken_at_ms, - now_ms, - ], - )?; - for item in items { - tx.execute( - "INSERT INTO mem_diff_snapshot_items (snapshot_id, item_id, title, content_hash, content, timestamp_ms, chunk_count) - VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)", - params![ - snapshot.id, - item.item_id, - item.title, - item.content_hash, - item.content, - item.timestamp_ms, - item.chunk_count, - ], - )?; - } - tx.commit()?; - Ok(()) -} - -pub fn list_snapshots( - conn: &Connection, - source_id: Option<&str>, - limit: u32, -) -> Result> { - let (sql, params_vec): (String, Vec>) = match source_id { - Some(sid) => ( - "SELECT id, source_id, source_kind, label, trigger_kind, item_count, taken_at_ms \ - FROM mem_diff_snapshots WHERE source_id = ?1 ORDER BY taken_at_ms DESC LIMIT ?2" - .to_string(), - vec![ - Box::new(sid.to_string()) as Box, - Box::new(limit), - ], - ), - None => ( - "SELECT id, source_id, source_kind, label, trigger_kind, item_count, taken_at_ms \ - FROM mem_diff_snapshots ORDER BY taken_at_ms DESC LIMIT ?1" - .to_string(), - vec![Box::new(limit) as Box], - ), - }; - let params_refs: Vec<&dyn rusqlite::ToSql> = params_vec.iter().map(|p| p.as_ref()).collect(); - let mut stmt = conn.prepare(&sql)?; - let rows = stmt.query_map(params_refs.as_slice(), |r| { - Ok(Snapshot { - id: r.get(0)?, - source_id: r.get(1)?, - source_kind: r.get(2)?, - label: r.get(3)?, - trigger: parse_trigger(r.get::<_, String>(4)?.as_str()), - item_count: r.get::<_, i64>(5)? as u32, - taken_at_ms: r.get(6)?, - }) - })?; - rows.collect::, _>>().map_err(Into::into) -} - -pub fn get_snapshot(conn: &Connection, snapshot_id: &str) -> Result> { - conn.query_row( - "SELECT id, source_id, source_kind, label, trigger_kind, item_count, taken_at_ms \ - FROM mem_diff_snapshots WHERE id = ?1", - [snapshot_id], - |r| { - Ok(Snapshot { - id: r.get(0)?, - source_id: r.get(1)?, - source_kind: r.get(2)?, - label: r.get(3)?, - trigger: parse_trigger(r.get::<_, String>(4)?.as_str()), - item_count: r.get::<_, i64>(5)? as u32, - taken_at_ms: r.get(6)?, - }) - }, - ) - .optional() - .map_err(Into::into) -} - -pub fn get_snapshot_items(conn: &Connection, snapshot_id: &str) -> Result> { - let mut stmt = conn.prepare( - "SELECT item_id, title, content_hash, content, timestamp_ms, chunk_count \ - FROM mem_diff_snapshot_items WHERE snapshot_id = ?1 ORDER BY item_id", - )?; - let rows = stmt.query_map([snapshot_id], |r| { - Ok(SnapshotItem { - item_id: r.get(0)?, - title: r.get(1)?, - content_hash: r.get(2)?, - content: r.get(3)?, - timestamp_ms: r.get(4)?, - chunk_count: r.get::<_, i64>(5)? as u32, - }) - })?; - rows.collect::, _>>().map_err(Into::into) -} - -pub fn latest_snapshots_for_source( - conn: &Connection, - source_id: &str, - count: u32, -) -> Result> { - list_snapshots(conn, Some(source_id), count) -} - -// ── Checkpoint CRUD ─────────────────────────────────────────────────── - -pub fn insert_checkpoint(conn: &Connection, checkpoint: &Checkpoint) -> Result<()> { - let tx = conn.unchecked_transaction()?; - tx.execute( - "INSERT INTO mem_diff_checkpoints (id, label, created_at_ms) VALUES (?1, ?2, ?3)", - params![checkpoint.id, checkpoint.label, checkpoint.created_at_ms], - )?; - for snap_id in &checkpoint.snapshot_ids { - tx.execute( - "INSERT INTO mem_diff_checkpoint_snapshots (checkpoint_id, snapshot_id) VALUES (?1, ?2)", - params![checkpoint.id, snap_id], - )?; - } - tx.commit()?; - Ok(()) -} - -pub fn list_checkpoints(conn: &Connection, limit: u32) -> Result> { - let mut stmt = conn.prepare( - "SELECT id, label, created_at_ms FROM mem_diff_checkpoints ORDER BY created_at_ms DESC LIMIT ?1", - )?; - let rows = stmt.query_map([limit], |r| { - Ok(( - r.get::<_, String>(0)?, - r.get::<_, String>(1)?, - r.get::<_, i64>(2)?, - )) - })?; - let mut checkpoints = Vec::new(); - for row in rows { - let (id, label, created_at_ms) = row?; - let snap_ids = checkpoint_snapshot_ids(conn, &id)?; - checkpoints.push(Checkpoint { - id, - label, - created_at_ms, - snapshot_ids: snap_ids, - }); - } - Ok(checkpoints) -} - -pub fn get_checkpoint(conn: &Connection, checkpoint_id: &str) -> Result> { - let row: Option<(String, String, i64)> = conn - .query_row( - "SELECT id, label, created_at_ms FROM mem_diff_checkpoints WHERE id = ?1", - [checkpoint_id], - |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)), - ) - .optional()?; - match row { - Some((id, label, created_at_ms)) => { - let snap_ids = checkpoint_snapshot_ids(conn, &id)?; - Ok(Some(Checkpoint { - id, - label, - created_at_ms, - snapshot_ids: snap_ids, - })) - } - None => Ok(None), - } -} - -fn checkpoint_snapshot_ids(conn: &Connection, checkpoint_id: &str) -> Result> { - let mut stmt = conn.prepare( - "SELECT snapshot_id FROM mem_diff_checkpoint_snapshots WHERE checkpoint_id = ?1", - )?; - let rows = stmt.query_map([checkpoint_id], |r| r.get(0))?; - rows.collect::, _>>().map_err(Into::into) -} - -// ── Read markers ────────────────────────────────────────────────────── - -/// Return the snapshot id a source's read marker currently points at, if any. -/// -/// The marker records the latest snapshot whose diff has been consumed by the -/// agent, so the next `diff_since_read` only surfaces newer changes. -pub fn get_read_marker(conn: &Connection, source_id: &str) -> Result> { - conn.query_row( - "SELECT snapshot_id FROM mem_diff_read_markers WHERE source_id = ?1", - [source_id], - |r| r.get::<_, String>(0), - ) - .optional() - .map_err(Into::into) -} - -/// Set (or advance) a source's read marker to a snapshot id. -pub fn upsert_read_marker( - conn: &Connection, - source_id: &str, - snapshot_id: &str, - marked_at_ms: i64, -) -> Result<()> { - conn.execute( - "INSERT INTO mem_diff_read_markers (source_id, snapshot_id, marked_at_ms) - VALUES (?1, ?2, ?3) - ON CONFLICT(source_id) DO UPDATE SET - snapshot_id = excluded.snapshot_id, - marked_at_ms = excluded.marked_at_ms", - params![source_id, snapshot_id, marked_at_ms], - )?; - Ok(()) -} - -// ── Cleanup ─────────────────────────────────────────────────────────── - -pub fn cleanup_old_snapshots( - conn: &Connection, - older_than_ms: i64, - max_per_source: u32, -) -> Result { - let age_deleted: i64 = conn.execute( - "DELETE FROM mem_diff_snapshots WHERE taken_at_ms < ?1", - params![older_than_ms], - )? as i64; - - let trim_deleted: i64 = conn.execute( - "DELETE FROM mem_diff_snapshots WHERE id IN ( - SELECT id FROM ( - SELECT id, ROW_NUMBER() OVER ( - PARTITION BY source_id ORDER BY taken_at_ms DESC - ) as rn FROM mem_diff_snapshots - ) WHERE rn > ?1 - )", - params![max_per_source], - )? as i64; - - Ok((age_deleted + trim_deleted).max(0) as u64) -} - -fn parse_trigger(s: &str) -> SnapshotTrigger { - match s { - "manual" => SnapshotTrigger::Manual, - _ => SnapshotTrigger::Auto, - } -} - -#[cfg(test)] -mod tests { - use super::*; - - fn test_conn() -> Connection { - let conn = Connection::open_in_memory().unwrap(); - conn.execute_batch("PRAGMA foreign_keys = ON;").unwrap(); - conn.execute_batch(SCHEMA_DDL).unwrap(); - conn - } - - fn test_snapshot(id: &str, source_id: &str) -> Snapshot { - Snapshot { - id: id.to_string(), - source_id: source_id.to_string(), - source_kind: "folder".to_string(), - label: "Test".to_string(), - trigger: SnapshotTrigger::Auto, - item_count: 2, - taken_at_ms: 1000, - } - } - - fn test_items() -> Vec { - vec![ - SnapshotItem { - item_id: "file_a".to_string(), - title: "File A".to_string(), - content_hash: "aaa".to_string(), - content: Some("alpha".to_string()), - timestamp_ms: Some(900), - chunk_count: 1, - }, - SnapshotItem { - item_id: "file_b".to_string(), - title: "File B".to_string(), - content_hash: "bbb".to_string(), - content: Some("beta".to_string()), - timestamp_ms: Some(950), - chunk_count: 2, - }, - ] - } - - #[test] - fn insert_and_retrieve_snapshot() { - let conn = test_conn(); - let snap = test_snapshot("snap_1", "src_x"); - let items = test_items(); - insert_snapshot(&conn, &snap, &items).unwrap(); - - let loaded = get_snapshot(&conn, "snap_1").unwrap().unwrap(); - assert_eq!(loaded.source_id, "src_x"); - assert_eq!(loaded.item_count, 2); - - let loaded_items = get_snapshot_items(&conn, "snap_1").unwrap(); - assert_eq!(loaded_items.len(), 2); - assert_eq!(loaded_items[0].item_id, "file_a"); - assert_eq!(loaded_items[0].content.as_deref(), Some("alpha")); - } - - #[test] - fn migration_adds_content_column_to_legacy_db() { - // Simulate a pre-existing diff.db whose snapshot-items table predates - // the `content` column (the original schema shape). - let conn = Connection::open_in_memory().unwrap(); - conn.execute_batch("PRAGMA foreign_keys = ON;").unwrap(); - conn.execute_batch( - "CREATE TABLE mem_diff_snapshot_items ( - snapshot_id TEXT NOT NULL, - item_id TEXT NOT NULL, - title TEXT NOT NULL DEFAULT '', - content_hash TEXT NOT NULL, - timestamp_ms INTEGER, - chunk_count INTEGER NOT NULL DEFAULT 1, - PRIMARY KEY (snapshot_id, item_id) - );", - ) - .unwrap(); - assert!(!column_exists(&conn, "mem_diff_snapshot_items", "content").unwrap()); - - // Re-running DDL + migrations must backfill the column. - conn.execute_batch(SCHEMA_DDL).unwrap(); - apply_column_migrations(&conn).unwrap(); - assert!(column_exists(&conn, "mem_diff_snapshot_items", "content").unwrap()); - - // Migration is idempotent on a second pass. - apply_column_migrations(&conn).unwrap(); - assert!(column_exists(&conn, "mem_diff_snapshot_items", "content").unwrap()); - } - - #[test] - fn read_marker_upsert_and_get() { - let conn = test_conn(); - // No marker initially. - assert_eq!(get_read_marker(&conn, "src_a").unwrap(), None); - - upsert_read_marker(&conn, "src_a", "snap_1", 1000).unwrap(); - assert_eq!( - get_read_marker(&conn, "src_a").unwrap().as_deref(), - Some("snap_1") - ); - - // Upsert advances the marker for the same source. - upsert_read_marker(&conn, "src_a", "snap_2", 2000).unwrap(); - assert_eq!( - get_read_marker(&conn, "src_a").unwrap().as_deref(), - Some("snap_2") - ); - - // Markers are per-source. - assert_eq!(get_read_marker(&conn, "src_b").unwrap(), None); - } - - #[test] - fn list_snapshots_filters_by_source() { - let conn = test_conn(); - insert_snapshot(&conn, &test_snapshot("s1", "src_a"), &[]).unwrap(); - insert_snapshot(&conn, &test_snapshot("s2", "src_b"), &[]).unwrap(); - - let all = list_snapshots(&conn, None, 100).unwrap(); - assert_eq!(all.len(), 2); - - let filtered = list_snapshots(&conn, Some("src_a"), 100).unwrap(); - assert_eq!(filtered.len(), 1); - assert_eq!(filtered[0].id, "s1"); - } - - #[test] - fn checkpoint_round_trip() { - let conn = test_conn(); - insert_snapshot(&conn, &test_snapshot("s1", "src_a"), &[]).unwrap(); - insert_snapshot(&conn, &test_snapshot("s2", "src_b"), &[]).unwrap(); - - let ckpt = Checkpoint { - id: "ckpt_1".to_string(), - label: "v1".to_string(), - created_at_ms: 2000, - snapshot_ids: vec!["s1".to_string(), "s2".to_string()], - }; - insert_checkpoint(&conn, &ckpt).unwrap(); - - let loaded = get_checkpoint(&conn, "ckpt_1").unwrap().unwrap(); - assert_eq!(loaded.label, "v1"); - assert_eq!(loaded.snapshot_ids.len(), 2); - - let all = list_checkpoints(&conn, 10).unwrap(); - assert_eq!(all.len(), 1); - } - - #[test] - fn cleanup_removes_old_snapshots() { - let conn = test_conn(); - let mut s1 = test_snapshot("s1", "src_a"); - s1.taken_at_ms = 100; - let mut s2 = test_snapshot("s2", "src_a"); - s2.taken_at_ms = 2000; - insert_snapshot(&conn, &s1, &test_items()).unwrap(); - insert_snapshot(&conn, &s2, &[]).unwrap(); - - let deleted = cleanup_old_snapshots(&conn, 500, 100).unwrap(); - assert_eq!(deleted, 1); - - let remaining = list_snapshots(&conn, None, 100).unwrap(); - assert_eq!(remaining.len(), 1); - assert_eq!(remaining[0].id, "s2"); - - // Items were cascade-deleted - let items = get_snapshot_items(&conn, "s1").unwrap(); - assert!(items.is_empty()); - } -} diff --git a/src/openhuman/memory_diff/tools.rs b/src/openhuman/memory_diff/tools.rs index 84731533c..f71ff2e44 100644 --- a/src/openhuman/memory_diff/tools.rs +++ b/src/openhuman/memory_diff/tools.rs @@ -136,19 +136,19 @@ impl Tool for MemoryDiffTool { .map(|s| (s.id.clone(), s.label.clone(), s.kind.as_str().to_string())) .collect(); - let counts: Vec<(String, String, String, usize)> = tokio::task::spawn_blocking(move || { - super::store::with_connection(&workspace_dir, |conn| { + let counts: Vec<(String, String, String, usize)> = + tokio::task::spawn_blocking(move || -> anyhow::Result<_> { + let ledger = super::git_store::Ledger::open(&workspace_dir)?; let mut out = Vec::new(); for (sid, label, kind) in &source_ids { - let snaps = super::store::list_snapshots(conn, Some(sid), 1000)?; - out.push((sid.clone(), label.clone(), kind.clone(), snaps.len())); + let count = ledger.snapshot_count_for_source(sid)?; + out.push((sid.clone(), label.clone(), kind.clone(), count)); } Ok(out) }) - }) - .await - .map_err(|e| anyhow::anyhow!("join: {e}"))? - .map_err(|e: anyhow::Error| anyhow::anyhow!("{e:#}"))?; + .await + .map_err(|e| anyhow::anyhow!("join: {e}"))? + .map_err(|e: anyhow::Error| anyhow::anyhow!("{e:#}"))?; let mut md = String::from("## Memory Sources (snapshot status)\n\n"); if counts.is_empty() { diff --git a/src/openhuman/memory_diff/types.rs b/src/openhuman/memory_diff/types.rs index 951d81e0d..3f1b83de7 100644 --- a/src/openhuman/memory_diff/types.rs +++ b/src/openhuman/memory_diff/types.rs @@ -30,23 +30,6 @@ pub struct Snapshot { pub taken_at_ms: i64, } -#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] -pub struct SnapshotItem { - pub item_id: String, - pub title: String, - pub content_hash: String, - /// Bounded copy of the item's concatenated content at snapshot time. - /// Persisted so text diffs can be computed against a later snapshot - /// even after the live chunk store has been overwritten by a new sync. - /// `None` for snapshots taken before content capture, or when the item - /// exceeded the storage bound. - #[serde(default, skip_serializing_if = "Option::is_none")] - pub content: Option, - #[serde(skip_serializing_if = "Option::is_none")] - pub timestamp_ms: Option, - pub chunk_count: u32, -} - #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] #[serde(rename_all = "snake_case")] pub enum ChangeKind {