From 948909a915a0ccfecfb5463b2eae417fca6585c6 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sat, 11 Jul 2026 18:32:12 +0000 Subject: [PATCH] =?UTF-8?q?refactor(tinycortex):=20W7=20=E2=80=94=20shim?= =?UTF-8?q?=20memory=5Fdiff=20over=20the=20crate=20DiffEngine?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Reduce memory_diff to a thin host shim over tinycortex::memory::diff: the snapshot/diff/checkpoint/ledger engine is now the crate's DiffEngine (a byte-identical port over the same /memory_diff/repo libgit2 layout — P9 parity: existing ledgers keep working unchanged). - ops.rs: the 9 async fns become thin spawn_blocking wrappers that build a DiffEngine + the host item-source seam and call the matching engine method, preserving the async + Result<_,String> signatures, DomainEvent publishes, and tracing that RPC/tools/sync/subconscious callers expect. - source.rs (new): ChunkStoreItemSource implements the crate's SnapshotItemSource seam by querying the authoritative mem_tree_chunks (the exact grouped/ordered query take_snapshot used before). It holds a source_id -> LIKE-prefix map (built from the full MemorySourceEntry list) because the Composio prefix (:%) isn't derivable from the logical id the crate passes. - types.rs: re-export the crate wire types (ChangeKind/Snapshot/DiffResult/ Checkpoint/CrossSourceDiff/ItemChange/DiffSummary/SnapshotTrigger). The old JsonSchema derive was vestigial — the RPC surface is hand-written TypeSchema::Ref schemas, not derived. - rpc.rs/tools.rs: repoint the direct Ledger::open list calls to the crate Ledger. - Delete git_store.rs (the whole libgit2 ledger engine — now the crate's). Parity note: the crate seam (items_for_source) has no Result channel, so a rare chunk-store read failure during snapshot yields an empty snapshot rather than the host's old propagated error. Self-healing (the ledger is a derived, rebuildable view; the next good snapshot restores state) and logged loudly. cargo check --lib: exit 0. Claude-Session: https://claude.ai/code/session_01X39btnEnHSTuPSYYvgyjrb --- src/openhuman/memory_diff/git_store.rs | 868 ------------------------- src/openhuman/memory_diff/mod.rs | 12 +- src/openhuman/memory_diff/ops.rs | 416 +++--------- src/openhuman/memory_diff/rpc.rs | 3 +- src/openhuman/memory_diff/source.rs | 198 ++++++ src/openhuman/memory_diff/tools.rs | 2 +- src/openhuman/memory_diff/types.rs | 139 +--- 7 files changed, 304 insertions(+), 1334 deletions(-) delete mode 100644 src/openhuman/memory_diff/git_store.rs create mode 100644 src/openhuman/memory_diff/source.rs diff --git a/src/openhuman/memory_diff/git_store.rs b/src/openhuman/memory_diff/git_store.rs deleted file mode 100644 index 86b471ee2f..0000000000 --- a/src/openhuman/memory_diff/git_store.rs +++ /dev/null @@ -1,868 +0,0 @@ -//! 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 827febbe6e..b2ba908d3b 100644 --- a/src/openhuman/memory_diff/mod.rs +++ b/src/openhuman/memory_diff/mod.rs @@ -11,8 +11,14 @@ //! 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. +//! the ledger is a derived view used purely for change tracking. +//! +//! W7: the snapshot/diff/checkpoint/ledger engine is now +//! `tinycortex::memory::diff::DiffEngine` (a byte-identical port over the same +//! `/memory_diff/repo` git layout). This module is a thin host shim: +//! [`ops`] async-wraps the engine, [`source`] supplies the chunk-store item +//! seam (`DiffEngine`'s `SnapshotItemSource`), [`types`] re-exports the crate +//! wire types, and [`rpc`]/[`schemas`]/[`tools`] keep the RPC + agent surface. //! //! Features: //! - Per-source snapshots (auto after sync, or manual via RPC) @@ -20,10 +26,10 @@ //! - 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 source; pub mod tools; pub mod types; diff --git a/src/openhuman/memory_diff/ops.rs b/src/openhuman/memory_diff/ops.rs index c4c9ddde9d..e85abf70de 100644 --- a/src/openhuman/memory_diff/ops.rs +++ b/src/openhuman/memory_diff/ops.rs @@ -1,82 +1,55 @@ -//! Business logic for memory diff: snapshot capture, diff computation, -//! checkpoints, and cleanup — all backed by the git ledger (`git_store`). +//! Business logic for memory diff — thin host async wrappers over +//! `tinycortex::memory::diff::DiffEngine` (W7). //! -//! `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 anyhow::{anyhow, bail}; +//! The snapshot/diff/checkpoint/ledger engine is the crate's; the git ledger it +//! writes lives at the same `/memory_diff/repo` path with the same +//! libgit2 layout, so existing ledgers keep working byte-for-byte. `DiffEngine` +//! is synchronous and generic over a chunk-source seam, so each op here builds +//! the host [`ChunkStoreItemSource`] (which reads the authoritative +//! `mem_tree_chunks`) and drives the engine inside `spawn_blocking`, preserving +//! the host's `async` + `Result<_, String>` signatures, the `DomainEvent` +//! publishes, and the tracing that RPC/tools/sync/subconscious callers expect. use crate::openhuman::config::Config; -use crate::openhuman::memory_sources::types::{MemorySourceEntry, SourceKind}; -use crate::openhuman::memory_store::chunks::store as chunk_store; +use crate::openhuman::memory_sources::types::MemorySourceEntry; + +use tinycortex::memory::diff::{DiffEngine, SourceDescriptor}; -use super::git_store::{Ledger, SnapshotMeta}; +use super::source::ChunkStoreItemSource; use super::types::*; +/// A crate [`SourceDescriptor`] from a host source entry. +fn descriptor(source: &MemorySourceEntry) -> SourceDescriptor { + SourceDescriptor::new( + source.id.clone(), + source.kind.as_str().to_string(), + source.label.clone(), + ) +} + /// Take a snapshot of the current chunk-store state for a source. /// -/// 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. +/// Reads from `mem_tree_chunks` (already-ingested data) via the item-source +/// seam, 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 prefix = source_id_prefix(source); - let config_clone = config.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 \ - FROM mem_tree_chunks \ - WHERE source_id LIKE ?1 \ - ORDER BY source_id, seq_in_source", - )?; - - let mut groups: HashMap> = HashMap::new(); - let rows = stmt.query_map([&prefix], |r| { - Ok((r.get::<_, String>(0)?, r.get::<_, String>(1)?)) - })?; - for row in rows { - let (composite_source_id, content) = row?; - let item_id = extract_item_id(&composite_source_id); - groups.entry(item_id).or_default().push(content); - } - - let mut items: Vec<(String, String)> = groups - .into_iter() - .map(|(item_id, parts)| (item_id, parts.join(""))) - .collect(); - 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 meta = SnapshotMeta { - source_id: source.id.clone(), - source_kind: source.kind.as_str().to_string(), - label: source.label.clone(), - trigger, - }; let workspace_dir = config.workspace_dir.clone(); - let now_ms = chrono::Utc::now().timestamp_millis(); + let config_clone = config.clone(); + let source_owned = source.clone(); + let desc = descriptor(source); let snapshot = tokio::task::spawn_blocking(move || -> anyhow::Result { - let ledger = Ledger::open(&workspace_dir)?; - ledger.commit_snapshot(&meta, &items, now_ms) + let items = ChunkStoreItemSource::single(config_clone, &source_owned); + let engine = DiffEngine::new(workspace_dir, items); + engine.take_snapshot(&desc, trigger) }) .await - .map_err(|e| format!("snapshot persist join: {e}"))? - .map_err(|e: anyhow::Error| format!("snapshot persist: {e:#}"))?; + .map_err(|e| format!("snapshot join error: {e}"))? + .map_err(|e: anyhow::Error| format!("take_snapshot: {e:#}"))?; tracing::debug!( snapshot_id = %snapshot.id, @@ -115,49 +88,13 @@ pub async fn compute_diff( include_text_diff: bool, ) -> Result { let workspace_dir = config.workspace_dir.clone(); + let config_clone = config.clone(); let to_id = to_snapshot_id.to_string(); let from_id = from_snapshot_id.map(|s| s.to_string()); 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 = 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 - ); - } - Some(s) - } - None => None, - }; - - 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, - }) + let engine = DiffEngine::new(workspace_dir, ChunkStoreItemSource::read_only(config_clone)); + engine.compute_diff(from_id.as_deref(), &to_id, include_text_diff) }) .await .map_err(|e| format!("diff join: {e}"))? @@ -171,29 +108,16 @@ pub async fn diff_since_last( include_text_diff: bool, ) -> Result { let workspace_dir = config.workspace_dir.clone(); + let config_clone = config.clone(); let source_id = source.id.clone(); - let snapshots = tokio::task::spawn_blocking(move || -> anyhow::Result> { - let ledger = Ledger::open(&workspace_dir)?; - ledger.latest_snapshots_for_source(&source_id, 2) + tokio::task::spawn_blocking(move || -> anyhow::Result { + let engine = DiffEngine::new(workspace_dir, ChunkStoreItemSource::read_only(config_clone)); + engine.diff_since_last(&source_id, include_text_diff) }) .await .map_err(|e| format!("diff_since_last join: {e}"))? - .map_err(|e: anyhow::Error| format!("diff_since_last: {e:#}"))?; - - match snapshots.len() { - 0 => Err("no snapshots found for this source".to_string()), - 1 => compute_diff(config, None, &snapshots[0].id, include_text_diff).await, - _ => { - compute_diff( - config, - Some(&snapshots[1].id), - &snapshots[0].id, - include_text_diff, - ) - .await - } - } + .map_err(|e: anyhow::Error| format!("diff_since_last: {e:#}")) } /// Diff a source's latest snapshot against its read marker — i.e. everything @@ -210,48 +134,21 @@ pub async fn diff_since_read( commit: bool, ) -> Result { let workspace_dir = config.workspace_dir.clone(); + let config_clone = config.clone(); let source_id = source.id.clone(); - // Resolve head (latest snapshot) and the marker's base snapshot. If the - // 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 = ledger.get_read_marker(&source_id)?; - let base_id = match marker { - Some(snap_id) if ledger.get_snapshot(&snap_id)?.is_some() => Some(snap_id), - _ => None, - }; - Ok((head, base_id)) - }, - ) + let diff = tokio::task::spawn_blocking(move || -> anyhow::Result { + let engine = DiffEngine::new(workspace_dir, ChunkStoreItemSource::read_only(config_clone)); + engine.diff_since_read(&source_id, include_text_diff, commit) + }) .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())?; - - 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(); - 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}"))? - .map_err(|e: anyhow::Error| format!("diff_since_read commit: {e:#}"))?; - tracing::debug!( source_id = %source.id, - snapshot_id = %head.id, + snapshot_id = %diff.to_snapshot_id, added = diff.summary.added, modified = diff.summary.modified, removed = diff.summary.removed, @@ -278,24 +175,23 @@ pub async fn mark_read(config: &Config, source_ids: Option>) -> Resu }; let workspace_dir = config.workspace_dir.clone(); + let config_clone = config.clone(); let ids_for_blocking = target_ids.clone(); + let (marked, snapshot_ids) = tokio::task::spawn_blocking(move || -> anyhow::Result<(u64, Vec)> { - let ledger = Ledger::open(&workspace_dir)?; - let mut count = 0u64; + let engine = + DiffEngine::new(workspace_dir, ChunkStoreItemSource::read_only(config_clone)); + // Gather the head snapshot ids that will be marked, for the event + // payload (the crate `mark_read` returns only a count). let mut snapshot_ids = Vec::new(); for sid in &ids_for_blocking { - if let Some(head) = ledger - .latest_snapshots_for_source(sid, 1)? - .into_iter() - .next() - { - ledger.set_read_marker(sid, &head.id)?; + if let Some(head) = engine.list_snapshots(Some(sid), 1)?.into_iter().next() { snapshot_ids.push(head.id); - count += 1; } } - Ok((count, snapshot_ids)) + let marked = engine.mark_read(&ids_for_blocking)?; + Ok((marked, snapshot_ids)) }) .await .map_err(|e| format!("mark_read join: {e}"))? @@ -317,66 +213,26 @@ pub async fn mark_read(config: &Config, source_ids: Option>) -> Resu } /// Create a checkpoint (git tag at HEAD) grouping the latest snapshot per -/// enabled source. +/// enabled source. Sources lacking a snapshot are baselined first. 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(); + let enabled: Vec = sources.into_iter().filter(|s| s.enabled).collect(); - // 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 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(lacking) - }) - .await - .map_err(|e| format!("checkpoint check join: {e}"))? - .map_err(|e: anyhow::Error| format!("checkpoint check: {e:#}"))?; - - 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 checkpoint_id = format!("ckpt_{}", uuid::Uuid::new_v4()); - let created_at_ms = chrono::Utc::now().timestamp_millis(); + let config_clone = config.clone(); 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, - }) + let descriptors: Vec = enabled.iter().map(descriptor).collect(); + let items = ChunkStoreItemSource::for_sources(config_clone, &enabled); + let engine = DiffEngine::new(workspace_dir, items); + engine.create_checkpoint(&label_owned, &descriptors) }) .await .map_err(|e| format!("checkpoint persist join: {e}"))? - .map_err(|e: anyhow::Error| format!("checkpoint persist: {e:#}"))?; + .map_err(|e: anyhow::Error| format!("create_checkpoint: {e:#}"))?; tracing::debug!( checkpoint_id = %checkpoint.id, @@ -394,61 +250,12 @@ pub async fn diff_since_checkpoint( include_text_diff: bool, ) -> Result { let workspace_dir = config.workspace_dir.clone(); + let config_clone = config.clone(); let ckpt_id = checkpoint_id.to_string(); - 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, - }) + let engine = DiffEngine::new(workspace_dir, ChunkStoreItemSource::read_only(config_clone)); + engine.diff_since_checkpoint(&ckpt_id, include_text_diff) }) .await .map_err(|e| format!("diff_since_checkpoint join: {e}"))? @@ -462,95 +269,21 @@ pub async fn diff_since_checkpoint( /// 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); + let config_clone = config.clone(); tokio::task::spawn_blocking(move || -> anyhow::Result { - let ledger = Ledger::open(&workspace_dir)?; - ledger.cleanup_checkpoints(cutoff) + let engine = DiffEngine::new(workspace_dir, ChunkStoreItemSource::read_only(config_clone)); + engine.cleanup(older_than_days) }) .await .map_err(|e| format!("cleanup join: {e}"))? .map_err(|e: anyhow::Error| format!("cleanup: {e:#}")) } -// ── Helpers ─────────────────────────────────────────────────────────── - -/// Build the `source_id LIKE` prefix that matches chunks belonging to a source. -/// Mirrors `memory_sources::status::source_id_prefix`. -fn source_id_prefix(source: &MemorySourceEntry) -> String { - match source.kind { - SourceKind::Composio => source - .toolkit - .as_deref() - .map(|t| format!("{t}:%")) - .unwrap_or_else(|| "__no_toolkit__:%".to_string()), - _ => format!("mem_src:{}:%", source.id), - } -} - -/// Extract the item-level id from a composite chunk source_id. -/// -/// For reader-backed: `mem_src:src_abc:readme.md` → `readme.md` -/// For Composio: `gmail:user@example.com:msg_xxx` → `user@example.com:msg_xxx` -fn extract_item_id(composite: &str) -> String { - if let Some(rest) = composite.strip_prefix("mem_src:") { - // Skip the source id segment - if let Some(pos) = rest.find(':') { - return rest[pos + 1..].to_string(); - } - } - // Composio or other: strip first segment - if let Some(pos) = composite.find(':') { - return composite[pos + 1..].to_string(); - } - composite.to_string() -} - #[cfg(test)] mod tests { use super::*; - use crate::openhuman::memory_diff::git_store::Ledger; - - #[test] - fn extract_item_id_reader_backed() { - assert_eq!(extract_item_id("mem_src:src_abc:readme.md"), "readme.md"); - assert_eq!( - extract_item_id("mem_src:src_abc:path/to/file.md"), - "path/to/file.md" - ); - } - - #[test] - fn extract_item_id_composio() { - assert_eq!( - extract_item_id("gmail:user@example.com:msg_xxx"), - "user@example.com:msg_xxx" - ); - } - - #[test] - fn extract_item_id_no_prefix() { - assert_eq!(extract_item_id("standalone"), "standalone"); - } - - #[test] - fn source_id_prefix_folder() { - assert_eq!( - source_id_prefix(&folder_source("src_abc")), - "mem_src:src_abc:%" - ); - } - - #[test] - 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 git ledger ──────────────── + use tinycortex::memory::diff::{Ledger, SnapshotMeta}; fn test_config() -> Config { let dir = tempfile::tempdir().unwrap(); @@ -564,7 +297,7 @@ mod tests { fn folder_source(id: &str) -> MemorySourceEntry { MemorySourceEntry { id: id.into(), - kind: SourceKind::Folder, + kind: crate::openhuman::memory_sources::types::SourceKind::Folder, label: "Docs".into(), enabled: true, toolkit: None, @@ -587,7 +320,8 @@ mod tests { } } - /// Seed a snapshot directly through the ledger (bypassing the chunk store). + /// Seed a snapshot directly through the (crate) ledger, bypassing the chunk + /// store — exercises the host async wrappers over real ledger state. fn seed( config: &Config, source_id: &str, diff --git a/src/openhuman/memory_diff/rpc.rs b/src/openhuman/memory_diff/rpc.rs index ae3d506e4d..b5b86ae684 100644 --- a/src/openhuman/memory_diff/rpc.rs +++ b/src/openhuman/memory_diff/rpc.rs @@ -6,7 +6,8 @@ use serde::{Deserialize, Serialize}; use crate::openhuman::config::rpc as config_rpc; use crate::rpc::RpcOutcome; -use super::git_store::Ledger; +use tinycortex::memory::diff::Ledger; + use super::ops; use super::types::*; diff --git a/src/openhuman/memory_diff/source.rs b/src/openhuman/memory_diff/source.rs new file mode 100644 index 0000000000..ee53ea3c4f --- /dev/null +++ b/src/openhuman/memory_diff/source.rs @@ -0,0 +1,198 @@ +//! The host implementation of the crate diff engine's chunk-source seam. +//! +//! `tinycortex::memory::diff::DiffEngine` is generic over a +//! [`SnapshotItemSource`](tinycortex::memory::diff::SnapshotItemSource): during +//! `take_snapshot` (directly, and transitively from `create_checkpoint` for any +//! source lacking a baseline) it asks the source for a source's already-ingested +//! items rather than re-calling readers. In OpenHuman that data lives in +//! `mem_tree_chunks`, so [`ChunkStoreItemSource`] answers the seam by querying +//! the chunk store — the exact query the host `take_snapshot` used before the +//! engine was ported to the crate (group by item id, concatenate chunk bodies in +//! `seq_in_source` order, sort by item id). +//! +//! ## Why the adapter holds a prefix map +//! +//! The crate calls `items_for_source(source_id)` with the *logical* source id, +//! but the host chunk `source_id LIKE` prefix is kind-dependent — Composio +//! sources key their chunks by `:%`, not `mem_src::%`, and the +//! toolkit is not derivable from the logical id alone. The adapter is therefore +//! built from the full [`MemorySourceEntry`] list (which carries `toolkit`) and +//! resolves each id → prefix up front. + +use std::collections::HashMap; + +use tinycortex::memory::diff::{extract_item_id, SnapshotItem, SnapshotItemSource}; + +use crate::openhuman::config::Config; +use crate::openhuman::memory_sources::types::{MemorySourceEntry, SourceKind}; + +/// Host [`SnapshotItemSource`] backed by `mem_tree_chunks`. +/// +/// Construct with [`single`](Self::single) for the per-source `take_snapshot` +/// path, [`for_sources`](Self::for_sources) for `create_checkpoint` (which may +/// baseline several sources), or [`read_only`](Self::read_only) for operations +/// that never materialise items (diff/list/cleanup) and just need *some* source +/// to satisfy the engine's type parameter. +pub struct ChunkStoreItemSource { + config: Config, + /// Logical source id → chunk `source_id LIKE` prefix. + prefixes: HashMap, +} + +impl ChunkStoreItemSource { + /// Adapter that can materialise items for any of `sources`. + pub fn for_sources(config: Config, sources: &[MemorySourceEntry]) -> Self { + let prefixes = sources + .iter() + .map(|s| (s.id.clone(), source_id_prefix(s))) + .collect(); + Self { config, prefixes } + } + + /// Adapter scoped to a single source (the common `take_snapshot` path). + pub fn single(config: Config, source: &MemorySourceEntry) -> Self { + let mut prefixes = HashMap::new(); + prefixes.insert(source.id.clone(), source_id_prefix(source)); + Self { config, prefixes } + } + + /// Adapter that never yields items — for read-only ops (`compute_diff`, + /// `diff_since_*`, `mark_read`, `diff_since_checkpoint`, `cleanup`) whose + /// engine calls only touch the ledger. `items_for_source` always returns + /// empty; it is never invoked on these paths. + pub fn read_only(config: Config) -> Self { + Self { + config, + prefixes: HashMap::new(), + } + } +} + +impl SnapshotItemSource for ChunkStoreItemSource { + fn items_for_source(&self, source_id: &str) -> Vec { + let Some(prefix) = self.prefixes.get(source_id) else { + return Vec::new(); + }; + + let result = + crate::openhuman::memory_store::chunks::store::with_connection(&self.config, |conn| { + let mut stmt = conn.prepare( + "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 rows = stmt.query_map([prefix], |r| { + Ok((r.get::<_, String>(0)?, r.get::<_, String>(1)?)) + })?; + for row in rows { + let (composite_source_id, content) = row?; + let item_id = extract_item_id(&composite_source_id); + groups.entry(item_id).or_default().push(content); + } + + let mut items: Vec = groups + .into_iter() + .map(|(item_id, parts)| SnapshotItem { + item_id, + content: parts.join(""), + }) + .collect(); + items.sort_by(|a, b| a.item_id.cmp(&b.item_id)); + Ok(items) + }); + + match result { + Ok(items) => items, + Err(e) => { + // The crate seam has no error channel. A chunk-store read + // failure here yields an empty snapshot (every item reads as + // removed for that one diff) rather than a propagated error — + // but the ledger is a derived, rebuildable view, so the next + // successful snapshot restores the true state. Log loudly. + tracing::error!( + source_id = %source_id, + error = %format!("{e:#}"), + "[memory_diff] chunk item-source query failed; snapshot will see no items" + ); + Vec::new() + } + } + } +} + +/// Build the `source_id LIKE` prefix that matches chunks belonging to a source. +/// Mirrors `memory_sources::status::source_id_prefix`. +pub(crate) fn source_id_prefix(source: &MemorySourceEntry) -> String { + match source.kind { + SourceKind::Composio => source + .toolkit + .as_deref() + .map(|t| format!("{t}:%")) + .unwrap_or_else(|| "__no_toolkit__:%".to_string()), + _ => format!("mem_src:{}:%", source.id), + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn folder_source(id: &str) -> MemorySourceEntry { + MemorySourceEntry { + id: id.into(), + kind: SourceKind::Folder, + label: "Docs".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, + } + } + + #[test] + fn source_id_prefix_folder() { + assert_eq!( + source_id_prefix(&folder_source("src_abc")), + "mem_src:src_abc:%" + ); + } + + #[test] + 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:%"); + } + + #[test] + fn source_id_prefix_composio_without_toolkit() { + let mut entry = folder_source("src_cmp"); + entry.kind = SourceKind::Composio; + entry.toolkit = None; + assert_eq!(source_id_prefix(&entry), "__no_toolkit__:%"); + } + + #[test] + fn read_only_adapter_never_yields_items() { + let source = ChunkStoreItemSource::read_only(Config::default()); + assert!(source.items_for_source("anything").is_empty()); + } +} diff --git a/src/openhuman/memory_diff/tools.rs b/src/openhuman/memory_diff/tools.rs index f71ff2e445..fc1cc62faf 100644 --- a/src/openhuman/memory_diff/tools.rs +++ b/src/openhuman/memory_diff/tools.rs @@ -138,7 +138,7 @@ impl Tool for MemoryDiffTool { let counts: Vec<(String, String, String, usize)> = tokio::task::spawn_blocking(move || -> anyhow::Result<_> { - let ledger = super::git_store::Ledger::open(&workspace_dir)?; + let ledger = tinycortex::memory::diff::Ledger::open(&workspace_dir)?; let mut out = Vec::new(); for (sid, label, kind) in &source_ids { let count = ledger.snapshot_count_for_source(sid)?; diff --git a/src/openhuman/memory_diff/types.rs b/src/openhuman/memory_diff/types.rs index 3f1b83de78..6459298b80 100644 --- a/src/openhuman/memory_diff/types.rs +++ b/src/openhuman/memory_diff/types.rs @@ -1,120 +1,19 @@ -//! Domain types for snapshot-based memory source change tracking. - -use schemars::JsonSchema; -use serde::{Deserialize, Serialize}; - -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] -#[serde(rename_all = "snake_case")] -pub enum SnapshotTrigger { - Auto, - Manual, -} - -impl SnapshotTrigger { - pub fn as_str(&self) -> &'static str { - match self { - Self::Auto => "auto", - Self::Manual => "manual", - } - } -} - -#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] -pub struct Snapshot { - pub id: String, - pub source_id: String, - pub source_kind: String, - pub label: String, - pub trigger: SnapshotTrigger, - pub item_count: u32, - pub taken_at_ms: i64, -} - -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] -#[serde(rename_all = "snake_case")] -pub enum ChangeKind { - Added, - Removed, - Modified, -} - -#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] -pub struct ItemChange { - pub item_id: String, - pub title: String, - pub kind: ChangeKind, - #[serde(skip_serializing_if = "Option::is_none")] - pub old_content_hash: Option, - #[serde(skip_serializing_if = "Option::is_none")] - pub new_content_hash: Option, - #[serde(skip_serializing_if = "Option::is_none")] - pub text_diff: Option, -} - -#[derive(Debug, Clone, Default, Serialize, Deserialize, JsonSchema)] -pub struct DiffSummary { - pub added: u32, - pub removed: u32, - pub modified: u32, - pub unchanged: u32, -} - -#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] -pub struct DiffResult { - pub source_id: String, - pub source_kind: String, - pub source_label: String, - pub from_snapshot_id: Option, - pub to_snapshot_id: String, - pub summary: DiffSummary, - pub changes: Vec, -} - -#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] -pub struct Checkpoint { - pub id: String, - pub label: String, - pub created_at_ms: i64, - pub snapshot_ids: Vec, -} - -#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] -pub struct CrossSourceDiff { - #[serde(skip_serializing_if = "Option::is_none")] - pub checkpoint_id: Option, - pub computed_at_ms: i64, - pub summary: DiffSummary, - pub per_source: Vec, -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn snapshot_trigger_round_trips() { - for trigger in [SnapshotTrigger::Auto, SnapshotTrigger::Manual] { - let json = serde_json::to_string(&trigger).unwrap(); - let decoded: SnapshotTrigger = serde_json::from_str(&json).unwrap(); - assert_eq!(decoded, trigger); - } - } - - #[test] - fn change_kind_round_trips() { - for kind in [ChangeKind::Added, ChangeKind::Removed, ChangeKind::Modified] { - let json = serde_json::to_string(&kind).unwrap(); - let decoded: ChangeKind = serde_json::from_str(&json).unwrap(); - assert_eq!(decoded, kind); - } - } - - #[test] - fn diff_summary_defaults_to_zero() { - let s = DiffSummary::default(); - assert_eq!(s.added, 0); - assert_eq!(s.removed, 0); - assert_eq!(s.modified, 0); - assert_eq!(s.unchanged, 0); - } -} +//! Domain types for snapshot-based memory-source change tracking — thin host +//! re-export of `tinycortex::memory::diff` types (W7). +//! +//! These are the published RPC/tool wire contract (serde `snake_case` enums + +//! stable field names). The crate port preserves them byte-for-byte, so the +//! host simply re-exports the crate types; the external consumers +//! (`memory_diff::rpc`/`tools`, `subconscious::profiles::memory`, and the RPC +//! controller schemas in `schemas.rs` which reference them by name) keep their +//! `memory_diff::types::*` import paths unchanged. +//! +//! Note: the host types formerly derived `schemars::JsonSchema`, but the RPC +//! surface is described by hand-written [`super::schemas`] (`TypeSchema::Ref` +//! strings), not derived schemas — so the derive was vestigial and its loss is +//! immaterial. + +pub use tinycortex::memory::diff::{ + ChangeKind, Checkpoint, CrossSourceDiff, DiffResult, DiffSummary, ItemChange, Snapshot, + SnapshotTrigger, +};