feat(memory_diff): back change ledger with git instead of SQLite (#4133)

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",

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"

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",

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 `<workspace>/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 `<source_id>/` 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_<uuid>` at HEAD
+//! - **Read marker**→ ref `refs/openhuman/read/<source_id>` → commit SHA
+//! - **Diff**       → `git diff <from-tree>..<to-tree>` 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<Self> {
+        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<Snapshot> {
+        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<Vec<Snapshot>> {
+        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<Option<Snapshot>> {
+        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<Vec<Snapshot>> {
+        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<usize> {
+        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<ItemChange>, 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<Option<String>> {
+        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<Option<Checkpoint>> {
+        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<Vec<Checkpoint>> {
+        let pattern = format!("{CHECKPOINT_PREFIX}*");
+        let names = self.repo.tag_names(Some(&pattern))?;
+        let mut out = Vec::new();
+        // StringArray::iter() yields Result<Option<&str>, _>; 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<u64> {
+        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::<i64>().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::<u32>().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<Signature<'static>> {
+    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<String, String> {
+    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<String> {
+    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<String> {
+    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");
+    }
+}

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 `<workspace>/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,
 };

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<Vec<Snapshot>> {
+        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<Vec<Checkpoint>> {
+        let ledger = Ledger::open(&workspace_dir)?;
+        ledger.list_checkpoints(limit)
     })
     .await
     .map_err(|e| format!("list_checkpoints join: {e}"))?

src/openhuman/memory_diff/store.rs

@@ -1,613 +0,0 @@
-//! SQLite persistence for memory diff snapshots and checkpoints.
-//!
-//! Own database at `<workspace>/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<T>(
-    workspace_dir: &Path,
-    f: impl FnOnce(&Connection) -> Result<T>,
-) -> Result<T> {
-    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<Connection> {
-    let mut last_err: Option<anyhow::Error> = 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<Connection> {
-    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<bool> {
-    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::<rusqlite::Error>() {
-        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<Vec<Snapshot>> {
-    let (sql, params_vec): (String, Vec<Box<dyn rusqlite::ToSql>>) = 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<dyn rusqlite::ToSql>,
-                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<dyn rusqlite::ToSql>],
-        ),
-    };
-    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::<Result<Vec<_>, _>>().map_err(Into::into)
-}
-
-pub fn get_snapshot(conn: &Connection, snapshot_id: &str) -> Result<Option<Snapshot>> {
-    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<Vec<SnapshotItem>> {
-    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::<Result<Vec<_>, _>>().map_err(Into::into)
-}
-
-pub fn latest_snapshots_for_source(
-    conn: &Connection,
-    source_id: &str,
-    count: u32,
-) -> Result<Vec<Snapshot>> {
-    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<Vec<Checkpoint>> {
-    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<Option<Checkpoint>> {
-    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<Vec<String>> {
-    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::<Result<Vec<String>, _>>().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<Option<String>> {
-    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<u64> {
-    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<SnapshotItem> {
-        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());
-    }
-}

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() {

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<String>,
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub timestamp_ms: Option<i64>,
-    pub chunk_count: u32,
-}
-
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
 #[serde(rename_all = "snake_case")]
 pub enum ChangeKind {