diff --git a/Cargo.lock b/Cargo.lock
index b0604e6f1..a3d54fcba 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -5219,6 +5219,7 @@ dependencies = [
  "sha1",
  "sha2 0.10.9",
  "shellexpand",
+ "similar",
  "socketioxide",
  "starship-battery",
  "sysinfo",
@@ -7453,6 +7454,12 @@ version = "0.1.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "e3a9fe34e3e7a50316060351f37187a3f546bce95496156754b601a5fa71b76e"
 
+[[package]]
+name = "similar"
+version = "2.7.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "bbbb5d9659141646ae647b42fe094daf6c6192d1620870b449d9557f748b2daa"
+
 [[package]]
 name = "siphasher"
 version = "1.0.3"
diff --git a/Cargo.toml b/Cargo.toml
index 22bb7d7a3..fb15c7695 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -68,6 +68,8 @@ argon2 = "0.5"
 rand = "0.10"
 dirs = "5"
 sha2 = "0.10"
+# Line-level text diffs for the memory_diff module (modified-item unified diffs).
+similar = "2"
 # Legacy SHA-1 only used for Tencent COS HMAC-SHA1 signing (yuanbao
 # channel media upload). Not used for any new security-sensitive work.
 sha1 = "0.10"
diff --git a/app/src-tauri/Cargo.lock b/app/src-tauri/Cargo.lock
index 40e8f5a00..83a37eb49 100644
--- a/app/src-tauri/Cargo.lock
+++ b/app/src-tauri/Cargo.lock
@@ -5437,6 +5437,7 @@ dependencies = [
  "sha1",
  "sha2 0.10.9",
  "shellexpand",
+ "similar",
  "socketioxide",
  "starship-battery",
  "sysinfo",
@@ -7764,6 +7765,12 @@ version = "0.3.9"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "703d5c7ef118737c72f1af64ad2f6f8c5e1921f818cdcb97b8fe6fc69bf66214"
 
+[[package]]
+name = "similar"
+version = "2.7.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "bbbb5d9659141646ae647b42fe094daf6c6192d1620870b449d9557f748b2daa"
+
 [[package]]
 name = "simplecss"
 version = "0.2.2"
diff --git a/src/core/event_bus/events.rs b/src/core/event_bus/events.rs
index 5d2f5219e..a14a1bd9f 100644
--- a/src/core/event_bus/events.rs
+++ b/src/core/event_bus/events.rs
@@ -285,6 +285,12 @@ pub enum DomainEvent {
         removed: usize,
         modified: usize,
     },
+    /// Read markers were committed for one or more sources, acknowledging
+    /// their current diffs as consumed.
+    MemoryDiffMarkedRead {
+        source_ids: Vec<String>,
+        snapshot_ids: Vec<String>,
+    },
 
     // ── Channels ────────────────────────────────────────────────────────
     /// An inbound channel message from the transport layer, ready for processing.
@@ -1172,7 +1178,8 @@ impl DomainEvent {
             | Self::MemoryIngestionCompleted { .. }
             | Self::DocumentCanonicalized { .. }
             | Self::MemoryDiffSnapshotTaken { .. }
-            | Self::MemoryDiffComputed { .. } => "memory",
+            | Self::MemoryDiffComputed { .. }
+            | Self::MemoryDiffMarkedRead { .. } => "memory",
 
             Self::CacheRebuilt { .. } => "learning",
 
@@ -1325,6 +1332,7 @@ impl DomainEvent {
             Self::DocumentCanonicalized { .. } => "DocumentCanonicalized",
             Self::MemoryDiffSnapshotTaken { .. } => "MemoryDiffSnapshotTaken",
             Self::MemoryDiffComputed { .. } => "MemoryDiffComputed",
+            Self::MemoryDiffMarkedRead { .. } => "MemoryDiffMarkedRead",
             Self::CacheRebuilt { .. } => "CacheRebuilt",
             Self::ChannelInboundMessage { .. } => "ChannelInboundMessage",
             Self::ChannelMessageReceived { .. } => "ChannelMessageReceived",
diff --git a/src/openhuman/memory_diff/ops.rs b/src/openhuman/memory_diff/ops.rs
index 75eab7fcd..040891ea9 100644
--- a/src/openhuman/memory_diff/ops.rs
+++ b/src/openhuman/memory_diff/ops.rs
@@ -15,6 +15,11 @@ use super::types::*;
 const DEFAULT_RETENTION_DAYS: u32 = 30;
 const MAX_SNAPSHOTS_PER_SOURCE: u32 = 100;
 const MAX_TEXT_DIFF_CHARS: usize = 2000;
+/// Upper bound on per-item content persisted into a snapshot, so the "from"
+/// side of a text diff survives the next sync overwriting the live chunk
+/// store. Items larger than this skip content capture (`content = None`);
+/// hash-based add/remove/modify detection still works for them.
+const MAX_SNAPSHOT_CONTENT_BYTES: usize = 64 * 1024;
 
 /// Take a snapshot of the current chunk-store state for a source.
 ///
@@ -61,10 +66,18 @@ pub async fn take_snapshot(
                 .map(|(item_id, acc)| {
                     let concat = acc.content_parts.join("");
                     let hash = sha256_hex(concat.as_bytes());
+                    let title = derive_title(&item_id, &concat);
+                    // Persist bounded content so a future diff has both sides.
+                    let content = if concat.len() <= MAX_SNAPSHOT_CONTENT_BYTES {
+                        Some(concat)
+                    } else {
+                        None
+                    };
                     SnapshotItem {
                         item_id,
-                        title: String::new(),
+                        title,
                         content_hash: hash,
+                        content,
                         timestamp_ms: acc.max_timestamp_ms,
                         chunk_count: acc.chunk_count,
                     }
@@ -239,9 +252,7 @@ pub async fn compute_diff(
             .collect();
 
         if !modified_ids.is_empty() {
-            let source_id = to_snap.source_id.clone();
-            let text_diffs =
-                compute_text_diffs_from_chunks(config, &source_id, &modified_ids).await;
+            let text_diffs = compute_text_diffs_from_snapshots(&from_map, &to_map, &modified_ids);
 
             for change in &mut changes {
                 if change.kind == ChangeKind::Modified {
@@ -297,6 +308,134 @@ pub async fn diff_since_last(
     }
 }
 
+/// Diff a source's latest snapshot against its read marker — i.e. everything
+/// that changed since the agent last *read* this source's diff.
+///
+/// When `commit` is true, the read marker is advanced to the head snapshot
+/// after the diff is computed, so a subsequent call returns only newer
+/// changes. This is the turn-to-turn primitive: read the world delta, then
+/// acknowledge it as consumed.
+pub async fn diff_since_read(
+    source: &MemorySourceEntry,
+    config: &Config,
+    include_text_diff: bool,
+    commit: bool,
+) -> Result<DiffResult, String> {
+    let workspace_dir = config.workspace_dir.clone();
+    let source_id = source.id.clone();
+
+    // Resolve head (latest snapshot) and the marker's base snapshot. If the
+    // marker points at a snapshot that has since been cleaned up, treat it as
+    // unread (base = None) rather than erroring.
+    let (head, base_id) = tokio::task::spawn_blocking(move || {
+        store::with_connection(&workspace_dir, |conn| {
+            let head = store::latest_snapshots_for_source(conn, &source_id, 1)?
+                .into_iter()
+                .next();
+            let marker = store::get_read_marker(conn, &source_id)?;
+            let base_id = match marker {
+                Some(snap_id) if store::get_snapshot(conn, &snap_id)?.is_some() => Some(snap_id),
+                _ => None,
+            };
+            Ok((head, base_id))
+        })
+    })
+    .await
+    .map_err(|e| format!("diff_since_read join: {e}"))?
+    .map_err(|e: anyhow::Error| format!("diff_since_read: {e:#}"))?;
+
+    let head = head.ok_or_else(|| "no snapshots found for this source".to_string())?;
+
+    // Marker already at head → nothing new since last read.
+    let from_id = match &base_id {
+        Some(id) if *id == head.id => Some(head.id.as_str()),
+        Some(id) => Some(id.as_str()),
+        None => None,
+    };
+
+    let diff = compute_diff(config, from_id, &head.id, include_text_diff).await?;
+
+    if commit {
+        let workspace_dir = config.workspace_dir.clone();
+        let source_id = source.id.clone();
+        let head_id = head.id.clone();
+        let now_ms = chrono::Utc::now().timestamp_millis();
+        tokio::task::spawn_blocking(move || {
+            store::with_connection(&workspace_dir, |conn| {
+                store::upsert_read_marker(conn, &source_id, &head_id, now_ms)
+            })
+        })
+        .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,
+            added = diff.summary.added,
+            modified = diff.summary.modified,
+            removed = diff.summary.removed,
+            "[memory_diff] read marker committed"
+        );
+    }
+
+    Ok(diff)
+}
+
+/// Commit a read marker for one or more sources, advancing each to its
+/// current head snapshot. When `source_ids` is `None`, marks all enabled
+/// sources that have at least one snapshot. Returns the number of markers set.
+pub async fn mark_read(config: &Config, source_ids: Option<Vec<String>>) -> Result<u64, String> {
+    let target_ids: Vec<String> = match source_ids {
+        Some(ids) => ids,
+        None => crate::openhuman::memory_sources::registry::list_sources()
+            .await
+            .map_err(|e| format!("list sources: {e}"))?
+            .into_iter()
+            .filter(|s| s.enabled)
+            .map(|s| s.id)
+            .collect(),
+    };
+
+    let workspace_dir = config.workspace_dir.clone();
+    let ids_for_blocking = target_ids.clone();
+    let (marked, snapshot_ids) = tokio::task::spawn_blocking(move || {
+        store::with_connection(&workspace_dir, |conn| {
+            let now_ms = chrono::Utc::now().timestamp_millis();
+            let mut count = 0u64;
+            let mut snapshot_ids = Vec::new();
+            for sid in &ids_for_blocking {
+                if let Some(head) = store::latest_snapshots_for_source(conn, sid, 1)?
+                    .into_iter()
+                    .next()
+                {
+                    store::upsert_read_marker(conn, sid, &head.id, now_ms)?;
+                    snapshot_ids.push(head.id);
+                    count += 1;
+                }
+            }
+            Ok((count, snapshot_ids))
+        })
+    })
+    .await
+    .map_err(|e| format!("mark_read join: {e}"))?
+    .map_err(|e: anyhow::Error| format!("mark_read: {e:#}"))?;
+
+    tracing::debug!(
+        sources = marked,
+        "[memory_diff] mark_read committed read markers"
+    );
+
+    crate::core::event_bus::publish_global(
+        crate::core::event_bus::DomainEvent::MemoryDiffMarkedRead {
+            source_ids: target_ids,
+            snapshot_ids,
+        },
+    );
+
+    Ok(marked)
+}
+
 /// Create a checkpoint that groups the latest snapshot per enabled source.
 pub async fn create_checkpoint(label: &str, config: &Config) -> Result<Checkpoint, String> {
     let sources = crate::openhuman::memory_sources::registry::list_sources()
@@ -511,19 +650,56 @@ fn truncate(s: &str, max_chars: usize) -> String {
     }
 }
 
-/// For modified items, read chunk content from the store and compute unified diffs.
-async fn compute_text_diffs_from_chunks(
-    config: &Config,
-    source_id: &str,
+/// Derive a human-readable title for an item from its content.
+///
+/// Uses the first non-empty line (a Markdown heading marker is stripped),
+/// trimmed and bounded. Falls back to the item id when no usable line exists
+/// (e.g. binary or empty content) so the diff output is never blank.
+fn derive_title(item_id: &str, content: &str) -> String {
+    let first_line = content
+        .lines()
+        .map(str::trim)
+        .find(|l| !l.is_empty())
+        .map(|l| l.trim_start_matches('#').trim());
+
+    match first_line {
+        Some(l) if !l.is_empty() => truncate(l, 120),
+        _ => item_id.to_string(),
+    }
+}
+
+/// Compute unified text diffs for modified items from the content stored in
+/// the two snapshots being compared. Both sides are read from the diff DB
+/// (bounded content captured at snapshot time), so this works even after the
+/// live chunk store has been overwritten by a later sync. Items whose content
+/// was too large to capture (`content = None` on either side) are skipped.
+fn compute_text_diffs_from_snapshots(
+    from_items: &HashMap<&str, &SnapshotItem>,
+    to_items: &HashMap<&str, &SnapshotItem>,
     item_ids: &[String],
 ) -> HashMap<String, String> {
-    // Text diffs require reading the current chunk content — this is already
-    // in the DB, not an API call. However, we only have the *current* content
-    // (the "to" side). The "from" side was overwritten by the new sync.
-    // For now, we note this limitation and return empty diffs.
-    // A future enhancement could store content snapshots or use the raw files.
-    let _ = (config, source_id, item_ids);
-    HashMap::new()
+    let mut out = HashMap::new();
+    for item_id in item_ids {
+        let (Some(from), Some(to)) = (
+            from_items.get(item_id.as_str()),
+            to_items.get(item_id.as_str()),
+        ) else {
+            continue;
+        };
+        let (Some(old), Some(new)) = (from.content.as_deref(), to.content.as_deref()) else {
+            continue;
+        };
+        let diff = similar::TextDiff::from_lines(old, new);
+        let unified = diff
+            .unified_diff()
+            .context_radius(3)
+            .header("before", "after")
+            .to_string();
+        if !unified.trim().is_empty() {
+            out.insert(item_id.clone(), unified);
+        }
+    }
+    out
 }
 
 #[derive(Default)]
@@ -635,4 +811,312 @@ mod tests {
         assert_eq!(h1, h2);
         assert_ne!(sha256_hex(b"hello"), sha256_hex(b"world"));
     }
+
+    #[test]
+    fn derive_title_uses_first_nonempty_line() {
+        assert_eq!(derive_title("file.md", "# Heading\nbody"), "Heading");
+        assert_eq!(
+            derive_title("file.md", "\n\n  Plain title  \nmore"),
+            "Plain title"
+        );
+    }
+
+    #[test]
+    fn derive_title_falls_back_to_item_id() {
+        assert_eq!(derive_title("doc_42", ""), "doc_42");
+        assert_eq!(derive_title("doc_42", "   \n  "), "doc_42");
+    }
+
+    // ── Integration-style ops tests over a temp diff.db ───────────────────
+
+    fn test_config() -> Config {
+        let dir = tempfile::tempdir().unwrap();
+        let mut config = Config::default();
+        config.workspace_dir = dir.path().to_path_buf();
+        // Leak the tempdir so the path stays valid for the test's lifetime.
+        std::mem::forget(dir);
+        config
+    }
+
+    fn item(item_id: &str, hash: &str, content: &str) -> SnapshotItem {
+        SnapshotItem {
+            item_id: item_id.to_string(),
+            title: item_id.to_string(),
+            content_hash: hash.to_string(),
+            content: Some(content.to_string()),
+            timestamp_ms: Some(1000),
+            chunk_count: 1,
+        }
+    }
+
+    fn seed(config: &Config, id: &str, source_id: &str, taken_at_ms: i64, items: &[SnapshotItem]) {
+        let snap = Snapshot {
+            id: id.to_string(),
+            source_id: source_id.to_string(),
+            source_kind: "folder".to_string(),
+            label: "Docs".to_string(),
+            trigger: SnapshotTrigger::Auto,
+            item_count: items.len() as u32,
+            taken_at_ms,
+        };
+        store::with_connection(&config.workspace_dir, |conn| {
+            store::insert_snapshot(conn, &snap, items)
+        })
+        .unwrap();
+    }
+
+    fn folder_source(id: &str) -> MemorySourceEntry {
+        MemorySourceEntry {
+            id: id.into(),
+            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,
+        }
+    }
+
+    #[tokio::test]
+    async fn compute_diff_detects_added_modified_removed() {
+        let config = test_config();
+        // from: a(h1), b(h2), c(h3)
+        seed(
+            &config,
+            "snap_from",
+            "src_a",
+            1000,
+            &[
+                item("a", "h1", "alpha"),
+                item("b", "h2", "beta"),
+                item("c", "h3", "gamma"),
+            ],
+        );
+        // to: a(h1, unchanged), b(h2b, modified), c removed, d(h4, added)
+        seed(
+            &config,
+            "snap_to",
+            "src_a",
+            2000,
+            &[
+                item("a", "h1", "alpha"),
+                item("b", "h2b", "beta v2"),
+                item("d", "h4", "delta"),
+            ],
+        );
+
+        let diff = compute_diff(&config, Some("snap_from"), "snap_to", false)
+            .await
+            .unwrap();
+
+        assert_eq!(diff.summary.added, 1, "d added");
+        assert_eq!(diff.summary.modified, 1, "b modified");
+        assert_eq!(diff.summary.removed, 1, "c removed");
+        assert_eq!(diff.summary.unchanged, 1, "a unchanged");
+
+        let kind_of = |id: &str| {
+            diff.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, "unchanged items are not in changes");
+    }
+
+    #[tokio::test]
+    async fn compute_diff_against_none_marks_all_added() {
+        let config = test_config();
+        seed(&config, "snap_to", "src_a", 1000, &[item("a", "h1", "x")]);
+        let diff = compute_diff(&config, None, "snap_to", false).await.unwrap();
+        assert_eq!(diff.summary.added, 1);
+        assert_eq!(diff.from_snapshot_id, None);
+    }
+
+    #[tokio::test]
+    async fn compute_diff_rejects_cross_source() {
+        let config = test_config();
+        seed(&config, "from_a", "src_a", 1000, &[]);
+        seed(&config, "to_b", "src_b", 2000, &[]);
+        let err = compute_diff(&config, Some("from_a"), "to_b", false)
+            .await
+            .unwrap_err();
+        assert!(err.contains("cross-source"), "got: {err}");
+    }
+
+    #[tokio::test]
+    async fn compute_diff_text_diff_only_when_requested() {
+        let config = test_config();
+        seed(
+            &config,
+            "f",
+            "src_a",
+            1000,
+            &[item("a", "h1", "line one\nline two\n")],
+        );
+        seed(
+            &config,
+            "t",
+            "src_a",
+            2000,
+            &[item("a", "h2", "line one\nline TWO changed\n")],
+        );
+
+        let without = compute_diff(&config, Some("f"), "t", false).await.unwrap();
+        assert!(without.changes[0].text_diff.is_none());
+
+        let with = compute_diff(&config, Some("f"), "t", true).await.unwrap();
+        let td = with.changes[0]
+            .text_diff
+            .as_ref()
+            .expect("text diff present");
+        assert!(td.contains("line TWO changed"), "got: {td}");
+    }
+
+    #[tokio::test]
+    async fn diff_since_last_handles_zero_one_two_snapshots() {
+        let config = test_config();
+        let source = folder_source("src_a");
+
+        // 0 snapshots → error
+        assert!(diff_since_last(&source, &config, false).await.is_err());
+
+        // 1 snapshot → everything added (diff vs None)
+        seed(&config, "s1", "src_a", 1000, &[item("a", "h1", "x")]);
+        let one = diff_since_last(&source, &config, false).await.unwrap();
+        assert_eq!(one.summary.added, 1);
+
+        // 2 snapshots → diff latest vs previous
+        seed(
+            &config,
+            "s2",
+            "src_a",
+            2000,
+            &[item("a", "h1", "x"), item("b", "h2", "y")],
+        );
+        let two = diff_since_last(&source, &config, false).await.unwrap();
+        assert_eq!(two.summary.added, 1, "b is new in s2");
+        assert_eq!(two.summary.unchanged, 1, "a unchanged");
+    }
+
+    #[tokio::test]
+    async fn diff_since_read_commits_marker_and_returns_only_new_changes() {
+        let config = test_config();
+        let source = folder_source("src_a");
+
+        seed(&config, "s1", "src_a", 1000, &[item("a", "h1", "x")]);
+
+        // First read: no marker → full diff (a added), and commit advances marker.
+        let first = diff_since_read(&source, &config, false, true)
+            .await
+            .unwrap();
+        assert_eq!(first.summary.added, 1);
+
+        // Second read with no new snapshot: marker == head → nothing changed.
+        let second = diff_since_read(&source, &config, false, true)
+            .await
+            .unwrap();
+        assert_eq!(second.summary.added, 0);
+        assert_eq!(second.summary.modified, 0);
+        assert_eq!(second.summary.removed, 0);
+        assert!(second.changes.is_empty());
+
+        // New snapshot then read: only the delta since the marker shows.
+        seed(
+            &config,
+            "s2",
+            "src_a",
+            2000,
+            &[item("a", "h1", "x"), item("b", "h2", "y")],
+        );
+        let third = diff_since_read(&source, &config, false, true)
+            .await
+            .unwrap();
+        assert_eq!(third.summary.added, 1, "only b is new since last read");
+        assert_eq!(third.summary.unchanged, 1);
+    }
+
+    #[tokio::test]
+    async fn diff_since_read_without_commit_does_not_advance_marker() {
+        let config = test_config();
+        let source = folder_source("src_a");
+        seed(&config, "s1", "src_a", 1000, &[item("a", "h1", "x")]);
+
+        // Preview (commit=false) twice → both show the full diff.
+        let a = diff_since_read(&source, &config, false, false)
+            .await
+            .unwrap();
+        let b = diff_since_read(&source, &config, false, false)
+            .await
+            .unwrap();
+        assert_eq!(a.summary.added, 1);
+        assert_eq!(b.summary.added, 1, "marker was not advanced");
+    }
+
+    #[tokio::test]
+    async fn mark_read_advances_marker_for_explicit_sources() {
+        let config = test_config();
+        let source = folder_source("src_a");
+        seed(&config, "s1", "src_a", 1000, &[item("a", "h1", "x")]);
+
+        let marked = mark_read(&config, Some(vec!["src_a".to_string()]))
+            .await
+            .unwrap();
+        assert_eq!(marked, 1);
+
+        // After marking, a read shows no changes (marker already at head).
+        let diff = diff_since_read(&source, &config, false, false)
+            .await
+            .unwrap();
+        assert_eq!(diff.summary.added, 0);
+        assert!(diff.changes.is_empty());
+    }
+
+    #[tokio::test]
+    async fn diff_since_checkpoint_aggregates_across_sources() {
+        let config = test_config();
+        // Baseline snapshots for two sources, grouped into a checkpoint.
+        seed(&config, "a1", "src_a", 1000, &[item("a", "h1", "x")]);
+        seed(&config, "b1", "src_b", 1000, &[item("b", "h1", "y")]);
+        let ckpt = Checkpoint {
+            id: "ckpt_1".to_string(),
+            label: "base".to_string(),
+            created_at_ms: 1500,
+            snapshot_ids: vec!["a1".to_string(), "b1".to_string()],
+        };
+        store::with_connection(&config.workspace_dir, |conn| {
+            store::insert_checkpoint(conn, &ckpt)
+        })
+        .unwrap();
+
+        // src_a gets a new head with a modification; src_b unchanged (no new head).
+        seed(&config, "a2", "src_a", 2000, &[item("a", "h2", "x v2")]);
+
+        let cross = diff_since_checkpoint("ckpt_1", &config, false)
+            .await
+            .unwrap();
+        assert_eq!(cross.summary.modified, 1, "src_a 'a' modified");
+        assert_eq!(
+            cross.per_source.len(),
+            1,
+            "only src_a changed; unchanged src_b is skipped"
+        );
+        assert_eq!(cross.per_source[0].source_id, "src_a");
+    }
 }
diff --git a/src/openhuman/memory_diff/rpc.rs b/src/openhuman/memory_diff/rpc.rs
index af3617154..2aa6c363d 100644
--- a/src/openhuman/memory_diff/rpc.rs
+++ b/src/openhuman/memory_diff/rpc.rs
@@ -61,6 +61,34 @@ pub struct DiffSinceLastResponse {
     pub diff: DiffResult,
 }
 
+#[derive(Debug, Deserialize)]
+pub struct DiffSinceReadRequest {
+    pub source_id: String,
+    #[serde(default)]
+    pub include_text_diff: Option<bool>,
+    /// Advance the read marker to the head snapshot after computing the diff.
+    /// Defaults to true so reading acknowledges the changes as consumed.
+    #[serde(default)]
+    pub commit: Option<bool>,
+}
+
+#[derive(Debug, Serialize)]
+pub struct DiffSinceReadResponse {
+    pub diff: DiffResult,
+}
+
+#[derive(Debug, Deserialize)]
+pub struct MarkReadRequest {
+    /// Sources to mark read. Omit to mark all enabled sources with a snapshot.
+    #[serde(default)]
+    pub source_ids: Option<Vec<String>>,
+}
+
+#[derive(Debug, Serialize)]
+pub struct MarkReadResponse {
+    pub marked: u64,
+}
+
 #[derive(Debug, Deserialize)]
 pub struct CreateCheckpointRequest {
     pub label: String,
@@ -195,6 +223,44 @@ pub async fn diff_since_last_rpc(
     Ok(RpcOutcome::new(DiffSinceLastResponse { diff }, vec![]))
 }
 
+pub async fn diff_since_read_rpc(
+    req: DiffSinceReadRequest,
+) -> Result<RpcOutcome<DiffSinceReadResponse>, String> {
+    let commit = req.commit.unwrap_or(true);
+    debug!(
+        "[memory_diff][rpc] diff_since_read source_id={} commit={}",
+        req.source_id, commit
+    );
+    let config = config_rpc::load_config_with_timeout().await?;
+    let source = crate::openhuman::memory_sources::get_source(&req.source_id)
+        .await?
+        .ok_or_else(|| format!("source not found: {}", req.source_id))?;
+
+    let diff = ops::diff_since_read(
+        &source,
+        &config,
+        req.include_text_diff.unwrap_or(false),
+        commit,
+    )
+    .await?;
+    debug!(
+        "[memory_diff][rpc] diff_since_read done added={} removed={} modified={}",
+        diff.summary.added, diff.summary.removed, diff.summary.modified
+    );
+    Ok(RpcOutcome::new(DiffSinceReadResponse { diff }, vec![]))
+}
+
+pub async fn mark_read_rpc(req: MarkReadRequest) -> Result<RpcOutcome<MarkReadResponse>, String> {
+    debug!(
+        "[memory_diff][rpc] mark_read source_ids={:?}",
+        req.source_ids
+    );
+    let config = config_rpc::load_config_with_timeout().await?;
+    let marked = ops::mark_read(&config, req.source_ids).await?;
+    debug!("[memory_diff][rpc] mark_read done marked={}", marked);
+    Ok(RpcOutcome::new(MarkReadResponse { marked }, vec![]))
+}
+
 pub async fn create_checkpoint_rpc(
     req: CreateCheckpointRequest,
 ) -> Result<RpcOutcome<CreateCheckpointResponse>, String> {
diff --git a/src/openhuman/memory_diff/schemas.rs b/src/openhuman/memory_diff/schemas.rs
index f6cd4c8c9..c5fa4eced 100644
--- a/src/openhuman/memory_diff/schemas.rs
+++ b/src/openhuman/memory_diff/schemas.rs
@@ -17,6 +17,8 @@ pub fn all_controller_schemas() -> Vec<ControllerSchema> {
         schemas("list_snapshots"),
         schemas("diff"),
         schemas("diff_since_last"),
+        schemas("diff_since_read"),
+        schemas("mark_read"),
         schemas("create_checkpoint"),
         schemas("list_checkpoints"),
         schemas("diff_since_checkpoint"),
@@ -42,6 +44,14 @@ pub fn all_registered_controllers() -> Vec<RegisteredController> {
             schema: schemas("diff_since_last"),
             handler: handle_diff_since_last,
         },
+        RegisteredController {
+            schema: schemas("diff_since_read"),
+            handler: handle_diff_since_read,
+        },
+        RegisteredController {
+            schema: schemas("mark_read"),
+            handler: handle_mark_read,
+        },
         RegisteredController {
             schema: schemas("create_checkpoint"),
             handler: handle_create_checkpoint,
@@ -163,6 +173,58 @@ fn schemas(function: &str) -> ControllerSchema {
                 required: true,
             }],
         },
+        "diff_since_read" => ControllerSchema {
+            namespace: NAMESPACE,
+            function: "diff_since_read",
+            description: "Diff a source's latest snapshot against the read marker — what \
+                          changed since the agent last read this source's diff. By default \
+                          commits the read marker so the next call returns only newer changes.",
+            inputs: vec![
+                FieldSchema {
+                    name: "source_id",
+                    ty: TypeSchema::String,
+                    comment: "Memory source id.",
+                    required: true,
+                },
+                FieldSchema {
+                    name: "include_text_diff",
+                    ty: TypeSchema::Option(Box::new(TypeSchema::Bool)),
+                    comment: "Include line-level text diffs for modified items.",
+                    required: false,
+                },
+                FieldSchema {
+                    name: "commit",
+                    ty: TypeSchema::Option(Box::new(TypeSchema::Bool)),
+                    comment: "Advance the read marker after diffing (default true). \
+                              Set false to preview without acknowledging.",
+                    required: false,
+                },
+            ],
+            outputs: vec![FieldSchema {
+                name: "diff",
+                ty: TypeSchema::Ref("DiffResult"),
+                comment: "Diff between the read marker and the latest snapshot.",
+                required: true,
+            }],
+        },
+        "mark_read" => ControllerSchema {
+            namespace: NAMESPACE,
+            function: "mark_read",
+            description: "Commit read markers, advancing each source to its current head \
+                          snapshot so prior changes are acknowledged as consumed.",
+            inputs: vec![FieldSchema {
+                name: "source_ids",
+                ty: TypeSchema::Option(Box::new(TypeSchema::Array(Box::new(TypeSchema::String)))),
+                comment: "Sources to mark read. Omit to mark all enabled sources with a snapshot.",
+                required: false,
+            }],
+            outputs: vec![FieldSchema {
+                name: "marked",
+                ty: TypeSchema::U64,
+                comment: "Number of read markers committed.",
+                required: true,
+            }],
+        },
         "create_checkpoint" => ControllerSchema {
             namespace: NAMESPACE,
             function: "create_checkpoint",
@@ -276,6 +338,20 @@ fn handle_diff_since_last(params: Map<String, Value>) -> ControllerFuture {
     })
 }
 
+fn handle_diff_since_read(params: Map<String, Value>) -> ControllerFuture {
+    Box::pin(async move {
+        let req = parse_value::<rpc::DiffSinceReadRequest>(Value::Object(params))?;
+        to_json(rpc::diff_since_read_rpc(req).await?)
+    })
+}
+
+fn handle_mark_read(params: Map<String, Value>) -> ControllerFuture {
+    Box::pin(async move {
+        let req = parse_value::<rpc::MarkReadRequest>(Value::Object(params))?;
+        to_json(rpc::mark_read_rpc(req).await?)
+    })
+}
+
 fn handle_create_checkpoint(params: Map<String, Value>) -> ControllerFuture {
     Box::pin(async move {
         let req = parse_value::<rpc::CreateCheckpointRequest>(Value::Object(params))?;
diff --git a/src/openhuman/memory_diff/store.rs b/src/openhuman/memory_diff/store.rs
index 022247a11..b67f0a83d 100644
--- a/src/openhuman/memory_diff/store.rs
+++ b/src/openhuman/memory_diff/store.rs
@@ -34,6 +34,7 @@ CREATE TABLE IF NOT EXISTS mem_diff_snapshot_items (
     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)
@@ -50,6 +51,12 @@ CREATE TABLE IF NOT EXISTS mem_diff_checkpoint_snapshots (
     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>(
@@ -102,9 +109,35 @@ fn open_and_init(db_path: &Path) -> Result<Connection> {
         .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") => {}
@@ -164,13 +197,14 @@ pub fn insert_snapshot(
     )?;
     for item in items {
         tx.execute(
-            "INSERT INTO mem_diff_snapshot_items (snapshot_id, item_id, title, content_hash, timestamp_ms, chunk_count)
-             VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
+            "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,
             ],
@@ -241,7 +275,7 @@ pub fn get_snapshot(conn: &Connection, snapshot_id: &str) -> Result<Option<Snaps
 
 pub fn get_snapshot_items(conn: &Connection, snapshot_id: &str) -> Result<Vec<SnapshotItem>> {
     let mut stmt = conn.prepare(
-        "SELECT item_id, title, content_hash, timestamp_ms, chunk_count \
+        "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| {
@@ -249,8 +283,9 @@ pub fn get_snapshot_items(conn: &Connection, snapshot_id: &str) -> Result<Vec<Sn
             item_id: r.get(0)?,
             title: r.get(1)?,
             content_hash: r.get(2)?,
-            timestamp_ms: r.get(3)?,
-            chunk_count: r.get::<_, i64>(4)? as u32,
+            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)
@@ -337,6 +372,40 @@ fn checkpoint_snapshot_ids(conn: &Connection, checkpoint_id: &str) -> Result<Vec
     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(
@@ -399,6 +468,7 @@ mod tests {
                 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,
             },
@@ -406,6 +476,7 @@ mod tests {
                 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,
             },
@@ -426,6 +497,60 @@ mod tests {
         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]
diff --git a/src/openhuman/memory_diff/tools.rs b/src/openhuman/memory_diff/tools.rs
index b66bbf03f..84731533c 100644
--- a/src/openhuman/memory_diff/tools.rs
+++ b/src/openhuman/memory_diff/tools.rs
@@ -22,8 +22,10 @@ impl Tool for MemoryDiffTool {
     }
 
     fn description(&self) -> &str {
-        "Check what changed in memory sources since the last sync or a named checkpoint. \
-         Returns a structured summary of added, removed, and modified items across one or all sources."
+        "Check what changed in memory sources since you last looked, the last sync, or a named \
+         checkpoint. Returns a structured summary of added, removed, and modified items. By \
+         default, reading a single source's diff commits a read marker so the next call only \
+         surfaces newer changes (set commit=false to preview without acknowledging)."
     }
 
     fn parameters_schema(&self) -> Value {
@@ -44,6 +46,18 @@ impl Tool for MemoryDiffTool {
                     "type": "boolean",
                     "description": "If true, include line-level text diffs for modified items (truncated).",
                     "default": false
+                },
+                "since_read": {
+                    "type": "boolean",
+                    "description": "When diffing a single source, show changes since you last read \
+                                    this source's diff (vs. since the previous sync). Default true.",
+                    "default": true
+                },
+                "commit": {
+                    "type": "boolean",
+                    "description": "When using since_read, advance the read marker so the next call \
+                                    only surfaces newer changes. Default true; set false to preview.",
+                    "default": true
                 }
             },
             "additionalProperties": false
@@ -51,6 +65,9 @@ impl Tool for MemoryDiffTool {
     }
 
     fn permission_level(&self) -> PermissionLevel {
+        // Read-only with respect to the user's data: the only write this tool
+        // performs is advancing the read marker in the module's own diff.db
+        // (internal bookkeeping under workspace state, never `action_dir`).
         PermissionLevel::ReadOnly
     }
 
@@ -61,10 +78,16 @@ impl Tool for MemoryDiffTool {
             .get("include_text_diff")
             .and_then(|v| v.as_bool())
             .unwrap_or(false);
+        let since_read = args
+            .get("since_read")
+            .and_then(|v| v.as_bool())
+            .unwrap_or(true);
+        let commit = args.get("commit").and_then(|v| v.as_bool()).unwrap_or(true);
 
         debug!(
-            "[memory_diff][tool] execute source_id={:?} checkpoint_id={:?} include_text_diff={}",
-            source_id, checkpoint_id, include_text_diff
+            "[memory_diff][tool] execute source_id={:?} checkpoint_id={:?} include_text_diff={} \
+             since_read={} commit={}",
+            source_id, checkpoint_id, include_text_diff, since_read, commit
         );
 
         let config = config_rpc::load_config_with_timeout()
@@ -87,9 +110,15 @@ impl Tool for MemoryDiffTool {
                 .map_err(|e| anyhow::anyhow!(e))?
                 .ok_or_else(|| anyhow::anyhow!("source not found: {sid}"))?;
 
-            let diff = ops::diff_since_last(&source, &config, include_text_diff)
-                .await
-                .map_err(|e| anyhow::anyhow!(e))?;
+            let diff = if since_read {
+                ops::diff_since_read(&source, &config, include_text_diff, commit)
+                    .await
+                    .map_err(|e| anyhow::anyhow!(e))?
+            } else {
+                ops::diff_since_last(&source, &config, include_text_diff)
+                    .await
+                    .map_err(|e| anyhow::anyhow!(e))?
+            };
             let md = format_diff_result(&diff);
             return Ok(ToolResult::success(md));
         }
@@ -254,3 +283,113 @@ fn format_cross_source_diff(diff: &CrossSourceDiff) -> String {
 
     md
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn change(item_id: &str, title: &str, kind: ChangeKind, text_diff: Option<&str>) -> ItemChange {
+        ItemChange {
+            item_id: item_id.to_string(),
+            title: title.to_string(),
+            kind,
+            old_content_hash: None,
+            new_content_hash: None,
+            text_diff: text_diff.map(str::to_string),
+        }
+    }
+
+    #[test]
+    fn format_diff_result_groups_changes_and_renders_text_diff() {
+        let diff = DiffResult {
+            source_id: "src_a".into(),
+            source_kind: "folder".into(),
+            source_label: "Docs".into(),
+            from_snapshot_id: Some("s1".into()),
+            to_snapshot_id: "s2".into(),
+            summary: DiffSummary {
+                added: 1,
+                removed: 1,
+                modified: 1,
+                unchanged: 2,
+            },
+            changes: vec![
+                change("new.md", "New Doc", ChangeKind::Added, None),
+                change(
+                    "edit.md",
+                    "Edited Doc",
+                    ChangeKind::Modified,
+                    Some("@@ -1 +1 @@\n-old\n+new"),
+                ),
+                // Empty title falls back to the item id.
+                change("gone.md", "", ChangeKind::Removed, None),
+            ],
+        };
+
+        let md = format_diff_result(&diff);
+        assert!(md.contains("1 added, 1 modified, 1 removed"));
+        assert!(md.contains("### Added\n- New Doc"));
+        assert!(md.contains("### Modified\n- Edited Doc"));
+        assert!(md.contains("```diff"), "text diff should be fenced: {md}");
+        assert!(md.contains("+new"));
+        assert!(
+            md.contains("### Removed\n- gone.md"),
+            "title falls back to id"
+        );
+    }
+
+    #[test]
+    fn format_diff_result_reports_no_changes() {
+        let diff = DiffResult {
+            source_id: "src_a".into(),
+            source_kind: "folder".into(),
+            source_label: "Docs".into(),
+            from_snapshot_id: Some("s1".into()),
+            to_snapshot_id: "s2".into(),
+            summary: DiffSummary::default(),
+            changes: vec![],
+        };
+        assert!(format_diff_result(&diff).contains("No changes detected."));
+    }
+
+    #[test]
+    fn format_cross_source_diff_breaks_down_per_source() {
+        let cross = CrossSourceDiff {
+            checkpoint_id: Some("ckpt_1".into()),
+            computed_at_ms: 0,
+            summary: DiffSummary {
+                added: 1,
+                modified: 0,
+                removed: 0,
+                unchanged: 0,
+            },
+            per_source: vec![DiffResult {
+                source_id: "src_a".into(),
+                source_kind: "folder".into(),
+                source_label: "Docs".into(),
+                from_snapshot_id: Some("s1".into()),
+                to_snapshot_id: "s2".into(),
+                summary: DiffSummary {
+                    added: 1,
+                    ..Default::default()
+                },
+                changes: vec![change("new.md", "New Doc", ChangeKind::Added, None)],
+            }],
+        };
+        let md = format_cross_source_diff(&cross);
+        assert!(md.contains("Total: 1 added"));
+        assert!(md.contains("### Docs (folder)"));
+        assert!(md.contains("+ New Doc"));
+    }
+
+    #[test]
+    fn format_cross_source_diff_empty_is_explicit() {
+        let cross = CrossSourceDiff {
+            checkpoint_id: Some("ckpt_1".into()),
+            computed_at_ms: 0,
+            summary: DiffSummary::default(),
+            per_source: vec![],
+        };
+        assert!(format_cross_source_diff(&cross).contains("No changes across any source"));
+    }
+}
diff --git a/src/openhuman/memory_diff/types.rs b/src/openhuman/memory_diff/types.rs
index 0121b2a8f..951d81e0d 100644
--- a/src/openhuman/memory_diff/types.rs
+++ b/src/openhuman/memory_diff/types.rs
@@ -35,6 +35,13 @@ 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,
diff --git a/tests/json_rpc_e2e.rs b/tests/json_rpc_e2e.rs
index 6b73d977a..9b4f88939 100644
--- a/tests/json_rpc_e2e.rs
+++ b/tests/json_rpc_e2e.rs
@@ -4232,6 +4232,235 @@ async fn json_rpc_memory_tree_end_to_end() {
     let _ = mock_join.await;
 }
 
+/// `openhuman.memory_diff_*` full lifecycle over JSON-RPC.
+///
+/// Drives the snapshot-based change tracker end to end: register a folder
+/// source, ingest chunks under its `mem_src:<id>:%` prefix across several
+/// snapshots, then exercise take_snapshot, diff_since_last, the read-marker
+/// watermark (diff_since_read + commit → empty on re-read → only-new after a
+/// later snapshot), mark_read, and cross-source checkpoints. This is the
+/// turn-to-turn world-diff assertion. (Per-item add/remove/modify detection
+/// and text diffs are covered exhaustively by the ops unit tests.)
+#[tokio::test]
+async fn json_rpc_memory_diff_snapshot_diff_and_read_marker_lifecycle() {
+    let _env_lock = json_rpc_e2e_env_lock();
+    let tmp = tempdir().expect("tempdir");
+    let home = tmp.path();
+    let openhuman_home = home.join(".openhuman");
+
+    let _home_guard = EnvVarGuard::set_to_path("HOME", home);
+    let _workspace_guard = EnvVarGuard::unset("OPENHUMAN_WORKSPACE");
+    let _backend_url_guard = EnvVarGuard::unset("BACKEND_URL");
+    let _vite_backend_guard = EnvVarGuard::unset("VITE_BACKEND_URL");
+    // Fall back to the inert (zero-vector) embedder; CI has no local Ollama.
+    let _embed_strict_guard = EnvVarGuard::set("OPENHUMAN_MEMORY_EMBED_STRICT", "false");
+    let _embed_endpoint_guard = EnvVarGuard::set("OPENHUMAN_MEMORY_EMBED_ENDPOINT", "");
+    let _embed_model_guard = EnvVarGuard::set("OPENHUMAN_MEMORY_EMBED_MODEL", "");
+
+    let (mock_addr, mock_join) = serve_on_ephemeral(mock_upstream_router()).await;
+    let mock_origin = format!("http://{}", mock_addr);
+    write_min_config(&openhuman_home, &mock_origin);
+
+    let (rpc_addr, rpc_join) = serve_on_ephemeral(build_core_http_router(false)).await;
+    let rpc_base = format!("http://{rpc_addr}");
+
+    tokio::time::sleep(Duration::from_millis(100)).await;
+
+    // ── Register a folder memory source; capture its generated id ─────────
+    let add = post_json_rpc(
+        &rpc_base,
+        9001,
+        "openhuman.memory_sources_add",
+        json!({
+            "kind": "folder",
+            "label": "Diff E2E Docs",
+            "path": home.join("docs").to_string_lossy(),
+        }),
+    )
+    .await;
+    let add_result = assert_no_jsonrpc_error(&add, "memory_sources_add");
+    let add_result = add_result.get("result").unwrap_or(add_result);
+    let source_id = add_result
+        .pointer("/source/id")
+        .and_then(Value::as_str)
+        .expect("source id")
+        .to_string();
+
+    // Chunks belonging to a reader-backed source live under `mem_src:<id>:%`.
+    let ingest = |id: i64, item: &str, body: &str| {
+        let composite = format!("mem_src:{source_id}:{item}");
+        post_json_rpc(
+            &rpc_base,
+            id,
+            "openhuman.memory_tree_ingest",
+            json!({
+                "source_kind": "document",
+                "source_id": composite,
+                "owner": "alice@example.com",
+                "payload": {
+                    "provider": "folder",
+                    "title": item,
+                    "body": body,
+                    "modified_at": 1700000000000_i64,
+                    "source_ref": format!("file://{item}"),
+                }
+            }),
+        )
+    };
+
+    // ── Generation 1: one doc, then snapshot ──────────────────────────────
+    let g1 = ingest(9002, "doc1.md", "First version of the launch plan.").await;
+    assert_no_jsonrpc_error(&g1, "ingest g1");
+
+    let snap1 = post_json_rpc(
+        &rpc_base,
+        9003,
+        "openhuman.memory_diff_take_snapshot",
+        json!({ "source_id": source_id }),
+    )
+    .await;
+    let snap1 = assert_no_jsonrpc_error(&snap1, "take_snapshot 1");
+    let snap1 = snap1.get("result").unwrap_or(snap1);
+    assert_eq!(
+        snap1.pointer("/snapshot/item_count"),
+        Some(&json!(1)),
+        "first snapshot has one item: {snap1}"
+    );
+
+    // ── Generation 2: add doc2, then snapshot ─────────────────────────────
+    let g2b = ingest(9005, "doc2.md", "Rollout checklist and staging notes.").await;
+    assert_no_jsonrpc_error(&g2b, "ingest g2b");
+
+    let snap2 = post_json_rpc(
+        &rpc_base,
+        9006,
+        "openhuman.memory_diff_take_snapshot",
+        json!({ "source_id": source_id }),
+    )
+    .await;
+    let snap2 = assert_no_jsonrpc_error(&snap2, "take_snapshot 2");
+    let snap2 = snap2.get("result").unwrap_or(snap2);
+    assert_eq!(
+        snap2.pointer("/snapshot/item_count"),
+        Some(&json!(2)),
+        "second snapshot has two items: {snap2}"
+    );
+
+    // ── diff_since_last: doc2 added, doc1 unchanged ───────────────────────
+    let dsl = post_json_rpc(
+        &rpc_base,
+        9007,
+        "openhuman.memory_diff_diff_since_last",
+        json!({ "source_id": source_id, "include_text_diff": true }),
+    )
+    .await;
+    let dsl = assert_no_jsonrpc_error(&dsl, "diff_since_last");
+    let dsl = dsl.get("result").unwrap_or(dsl);
+    assert_eq!(dsl.pointer("/diff/summary/added"), Some(&json!(1)), "{dsl}");
+    assert_eq!(
+        dsl.pointer("/diff/summary/unchanged"),
+        Some(&json!(1)),
+        "{dsl}"
+    );
+
+    // ── diff_since_read (commit) then re-read → empty (marker advanced) ───
+    let read1 = post_json_rpc(
+        &rpc_base,
+        9008,
+        "openhuman.memory_diff_diff_since_read",
+        json!({ "source_id": source_id }),
+    )
+    .await;
+    let read1 = assert_no_jsonrpc_error(&read1, "diff_since_read 1");
+    let read1 = read1.get("result").unwrap_or(read1);
+    // First read with no prior marker reports everything as added.
+    assert_eq!(
+        read1.pointer("/diff/summary/added"),
+        Some(&json!(2)),
+        "{read1}"
+    );
+
+    let read2 = post_json_rpc(
+        &rpc_base,
+        9009,
+        "openhuman.memory_diff_diff_since_read",
+        json!({ "source_id": source_id }),
+    )
+    .await;
+    let read2 = assert_no_jsonrpc_error(&read2, "diff_since_read 2");
+    let read2 = read2.get("result").unwrap_or(read2);
+    assert_eq!(
+        read2.pointer("/diff/summary/added"),
+        Some(&json!(0)),
+        "{read2}"
+    );
+    assert_eq!(
+        read2.pointer("/diff/summary/modified"),
+        Some(&json!(0)),
+        "second read after commit is empty: {read2}"
+    );
+
+    // ── mark_read is idempotent on an already-acknowledged source ─────────
+    let mark = post_json_rpc(
+        &rpc_base,
+        9010,
+        "openhuman.memory_diff_mark_read",
+        json!({ "source_ids": [source_id] }),
+    )
+    .await;
+    let mark = assert_no_jsonrpc_error(&mark, "mark_read");
+    let mark = mark.get("result").unwrap_or(mark);
+    assert_eq!(mark.get("marked"), Some(&json!(1)), "{mark}");
+
+    // ── Checkpoint + cross-source diff after a further change ─────────────
+    let ckpt = post_json_rpc(
+        &rpc_base,
+        9011,
+        "openhuman.memory_diff_create_checkpoint",
+        json!({ "label": "baseline" }),
+    )
+    .await;
+    let ckpt = assert_no_jsonrpc_error(&ckpt, "create_checkpoint");
+    let ckpt = ckpt.get("result").unwrap_or(ckpt);
+    let checkpoint_id = ckpt
+        .pointer("/checkpoint/id")
+        .and_then(Value::as_str)
+        .expect("checkpoint id")
+        .to_string();
+
+    // Add doc3, snapshot, then diff since the checkpoint.
+    let g3 = ingest(9012, "doc3.md", "Post-launch retro and metrics.").await;
+    assert_no_jsonrpc_error(&g3, "ingest g3");
+    let snap3 = post_json_rpc(
+        &rpc_base,
+        9013,
+        "openhuman.memory_diff_take_snapshot",
+        json!({ "source_id": source_id }),
+    )
+    .await;
+    assert_no_jsonrpc_error(&snap3, "take_snapshot 3");
+
+    let since_ckpt = post_json_rpc(
+        &rpc_base,
+        9014,
+        "openhuman.memory_diff_diff_since_checkpoint",
+        json!({ "checkpoint_id": checkpoint_id }),
+    )
+    .await;
+    let since_ckpt = assert_no_jsonrpc_error(&since_ckpt, "diff_since_checkpoint");
+    let since_ckpt = since_ckpt.get("result").unwrap_or(since_ckpt);
+    assert_eq!(
+        since_ckpt.pointer("/diff/summary/added"),
+        Some(&json!(1)),
+        "doc3 added since checkpoint: {since_ckpt}"
+    );
+
+    rpc_join.abort();
+    let _ = rpc_join.await;
+    mock_join.abort();
+    let _ = mock_join.await;
+}
+
 /// `openhuman.memory_tree_cover_window` over RPC: ingest a chunk, then assert
 /// the windowed minimum-cover returns it raw inside the window and nothing
 /// outside it. One ingested chunk doesn't reach the seal fanout, so this also