mirror of
https://github.com/tinyhumansai/openhuman.git
synced 2026-07-27 21:08:00 +00:00
feat(memory_diff): read-marker watermark, real text diffs, item titles (#4100)
This commit is contained in:
Steven Enamakel
Generated
+7
@@ -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"
|
||||
|
||||
@@ -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"
|
||||
|
||||
Generated
+7
@@ -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"
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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");
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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> {
|
||||
|
||||
@@ -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))?;
|
||||
|
||||
@@ -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]
|
||||
|
||||
@@ -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"));
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user