From 6395f642e10b90a8a68c2bbb89bbfb92838638fa Mon Sep 17 00:00:00 2001 From: Steven Enamakel <31011319+senamakel@users.noreply.github.com> Date: Sun, 28 Jun 2026 21:24:07 -0700 Subject: [PATCH] feat(memory): track summary-only wiki git history (#4168) --- src/openhuman/memory_store/content/README.md | 1 + src/openhuman/memory_store/content/mod.rs | 1 + .../memory_store/content/wiki_git/mod.rs | 364 ++++++++++++++++++ .../memory_store/content/wiki_git/tests.rs | 328 ++++++++++++++++ src/openhuman/memory_tree/ingest.rs | 34 ++ src/openhuman/memory_tree/tree/bucket_seal.rs | 93 ++++- .../memory_tree/tree/bucket_seal_tests.rs | 1 + tests/memory_artifacts_e2e.rs | 95 +++++ 8 files changed, 915 insertions(+), 2 deletions(-) create mode 100644 src/openhuman/memory_store/content/wiki_git/mod.rs create mode 100644 src/openhuman/memory_store/content/wiki_git/tests.rs diff --git a/src/openhuman/memory_store/content/README.md b/src/openhuman/memory_store/content/README.md index 74e36d79d..36d019ec5 100644 --- a/src/openhuman/memory_store/content/README.md +++ b/src/openhuman/memory_store/content/README.md @@ -14,6 +14,7 @@ The body is **immutable** once written — only the YAML front-matter `tags:` bl - [`raw.rs`](raw.rs) — verbatim source-byte mirror under `/raw/`. Writes the unmodified upstream payload (eml, slack json, raw markdown) so downstream callers can re-canonicalise without re-fetching. - [`obsidian.rs`](obsidian.rs) + [`obsidian_defaults/`](obsidian_defaults/) — bootstrap an `.obsidian/` config (workspace, graph, app) into the content root on first write so a user opening the vault gets a usable view. - [`tags.rs`](tags.rs) — post-extraction tag rewrites. `update_chunk_tags` (atomic tempfile rewrite of the `tags:` block) and `update_summary_tags` (fetches entities from `mem_tree_entity_index`, builds Obsidian `kind/Value` tags, rewrites, verifies body SHA is unchanged). `slugify_tag_kind`, `slugify_tag_value`, `entity_tag` build the tag strings. +- [`wiki_git/`](wiki_git/) — initializes `/wiki/.git`, commits only summary-node markdown under `summaries/**` plus the repo `.gitignore`, and stores read high-water marks as lightweight `refs/tags/read/*` pointers. Summary files are staged by `atomic.rs`; seal/ingest callers create descriptive git commits after SQLite persistence succeeds. ## Integrity contract diff --git a/src/openhuman/memory_store/content/mod.rs b/src/openhuman/memory_store/content/mod.rs index d88bd4287..302a76787 100644 --- a/src/openhuman/memory_store/content/mod.rs +++ b/src/openhuman/memory_store/content/mod.rs @@ -20,6 +20,7 @@ pub mod paths; pub mod raw; pub mod read; pub mod tags; +pub mod wiki_git; use std::path::Path; diff --git a/src/openhuman/memory_store/content/wiki_git/mod.rs b/src/openhuman/memory_store/content/wiki_git/mod.rs new file mode 100644 index 000000000..176678478 --- /dev/null +++ b/src/openhuman/memory_store/content/wiki_git/mod.rs @@ -0,0 +1,364 @@ +//! Git history for derived wiki summary nodes. +//! +//! The repository lives at `/wiki/.git` and intentionally tracks +//! only summary-node markdown (`summaries/**`) plus its own restrictive +//! `.gitignore`. Raw source mirrors, chunk intermediates, Obsidian defaults, +//! and future non-summary wiki artifacts are left out of history. + +use std::path::{Path, PathBuf}; +use std::sync::Mutex; + +use anyhow::{Context, Result}; +use chrono::{DateTime, Utc}; +use git2::{ErrorCode, Oid, Repository, RepositoryOpenFlags, Signature}; + +use super::paths::WIKI_PREFIX; + +static WIKI_GIT_LOCK: Mutex<()> = Mutex::new(()); + +const SIG_NAME: &str = "OpenHuman Memory"; +const SIG_EMAIL: &str = "memory-wiki@openhuman.local"; +const GITIGNORE_BODY: &str = "*\n!/.gitignore\n!/summaries/\n!/summaries/**\n"; + +/// Metadata for one summary node included in a wiki git commit. +#[derive(Clone, Debug)] +pub struct SummaryCommitEntry { + pub summary_id: String, + pub content_path: String, + pub level: u32, + pub child_count: usize, + pub token_count: u32, + pub time_range_start: DateTime, + pub time_range_end: DateTime, +} + +/// Metadata for one tree seal represented as a wiki git commit. +#[derive(Clone, Debug)] +pub struct SummaryCommitBatch { + pub reason: String, + pub tree_id: String, + pub tree_scope: String, + pub entries: Vec, +} + +/// Ensure the wiki repository exists and has a commit containing the supplied +/// summary files. Existing non-summary tracked entries are removed from the +/// index so history stays scoped to summary nodes only. +pub fn commit_summaries(content_root: &Path, batch: &SummaryCommitBatch) -> Result<()> { + if batch.entries.is_empty() { + return Ok(()); + } + let summary_repo_paths: Vec = batch + .entries + .iter() + .map(|entry| summary_repo_path(&entry.content_path)) + .collect::>>()?; + let _guard = WIKI_GIT_LOCK.lock().expect("memory wiki git lock poisoned"); + + let repo = open_prepared_repo(content_root)?; + let wiki_root = content_root.join(WIKI_PREFIX); + + let mut index = repo.index().context("open wiki git index")?; + prune_stale_or_non_summary_entries(&mut index, repo.workdir().unwrap_or(&wiki_root))?; + index + .add_path(Path::new(".gitignore")) + .context("stage wiki .gitignore")?; + for path in &summary_repo_paths { + index + .add_path(Path::new(path)) + .with_context(|| format!("stage wiki summary: {path}"))?; + } + stage_existing_summary_paths(&mut index, &wiki_root)?; + index + .write() + .context("write wiki git index after staging summary")?; + + commit_index_if_changed(&repo, batch) +} + +/// Add a timestamped lightweight git tag that represents a reader's high-water +/// mark, and move a stable `latest` alias for quick lookup. +/// +/// This writes `refs/tags/read//` +/// to `target_commit`, or to wiki `HEAD` when `target_commit` is `None`, and +/// also updates `refs/tags/read//latest`. Tags update read +/// state without creating another history commit. +pub fn set_read_pointer_tag( + content_root: &Path, + pointer_id: &str, + target_commit: Option<&str>, +) -> Result { + let _guard = WIKI_GIT_LOCK.lock().expect("memory wiki git lock poisoned"); + let repo = open_prepared_repo(content_root)?; + let oid = match target_commit { + Some(commit) => { + Oid::from_str(commit).with_context(|| format!("bad commit id: {commit}"))? + } + None => repo.head()?.peel_to_commit()?.id(), + }; + let tag_ref = read_pointer_timestamp_ref(pointer_id, Utc::now()); + repo.reference(&tag_ref, oid, true, "advance memory wiki read pointer") + .with_context(|| format!("set wiki read pointer tag: {tag_ref}"))?; + let latest_ref = read_pointer_latest_ref(pointer_id); + repo.reference( + &latest_ref, + oid, + true, + "advance latest memory wiki read pointer", + ) + .with_context(|| format!("set latest wiki read pointer tag: {latest_ref}"))?; + log::debug!( + "[content_store::wiki_git] advanced read pointer tags {} latest={} -> {}", + tag_ref, + latest_ref, + oid + ); + Ok(oid.to_string()) +} + +/// Return the commit id a read-pointer tag currently references. +pub fn get_read_pointer_tag(content_root: &Path, pointer_id: &str) -> Result> { + let _guard = WIKI_GIT_LOCK.lock().expect("memory wiki git lock poisoned"); + let wiki_root = content_root.join(WIKI_PREFIX); + let repo = match open_existing_repo(&wiki_root) { + Ok(repo) => repo, + Err(err) if err.code() == ErrorCode::NotFound => return Ok(None), + Err(err) => return Err(err).context("open wiki git repo for read pointer"), + }; + let tag_ref = read_pointer_latest_ref(pointer_id); + let target = match repo.find_reference(&tag_ref) { + Ok(reference) => Ok(reference.target().map(|oid| oid.to_string())), + Err(err) if err.code() == ErrorCode::NotFound => Ok(None), + Err(err) => Err(err).with_context(|| format!("find wiki read pointer tag: {tag_ref}")), + }; + target +} + +fn open_prepared_repo(content_root: &Path) -> Result { + let wiki_root = content_root.join(WIKI_PREFIX); + std::fs::create_dir_all(&wiki_root) + .with_context(|| format!("create wiki git root: {}", wiki_root.display()))?; + + let repo = open_or_init_repo(&wiki_root)?; + ensure_gitignore(&wiki_root)?; + Ok(repo) +} + +fn open_or_init_repo(wiki_root: &Path) -> Result { + match open_existing_repo(wiki_root) { + Ok(repo) => Ok(repo), + Err(err) if err.code() == ErrorCode::NotFound => { + log::debug!( + "[content_store::wiki_git] initialising summary wiki git repo at {}", + wiki_root.display() + ); + Repository::init(wiki_root) + .with_context(|| format!("init wiki git repo: {}", wiki_root.display())) + } + Err(err) => { + Err(err).with_context(|| format!("open wiki git repo: {}", wiki_root.display())) + } + } +} + +fn open_existing_repo(wiki_root: &Path) -> Result { + Repository::open_ext( + wiki_root, + RepositoryOpenFlags::NO_SEARCH, + &[] as &[&std::ffi::OsStr], + ) +} + +fn ensure_gitignore(wiki_root: &Path) -> Result<()> { + let path = wiki_root.join(".gitignore"); + match std::fs::read_to_string(&path) { + Ok(existing) if existing == GITIGNORE_BODY => Ok(()), + _ => { + std::fs::write(&path, GITIGNORE_BODY) + .with_context(|| format!("write wiki gitignore: {}", path.display()))?; + log::debug!( + "[content_store::wiki_git] wrote summary-only .gitignore at {}", + path.display() + ); + Ok(()) + } + } +} + +fn prune_stale_or_non_summary_entries(index: &mut git2::Index, wiki_root: &Path) -> Result<()> { + let to_remove: Vec = index + .iter() + .filter_map(|entry| { + let path = std::str::from_utf8(&entry.path).ok()?; + if should_keep_index_entry(wiki_root, path) { + None + } else { + Some(PathBuf::from(path)) + } + }) + .collect(); + + for path in to_remove { + index + .remove_path(&path) + .with_context(|| format!("remove non-summary wiki git entry: {}", path.display()))?; + } + Ok(()) +} + +fn should_keep_index_entry(wiki_root: &Path, path: &str) -> bool { + if !is_tracked_wiki_path(path) { + return false; + } + path == ".gitignore" || wiki_root.join(path).exists() +} + +fn is_tracked_wiki_path(path: &str) -> bool { + path == ".gitignore" || path.starts_with("summaries/") +} + +fn stage_existing_summary_paths(index: &mut git2::Index, wiki_root: &Path) -> Result<()> { + let summaries_root = wiki_root.join("summaries"); + if !summaries_root.exists() { + return Ok(()); + } + stage_summary_dir(index, wiki_root, &summaries_root) +} + +fn stage_summary_dir(index: &mut git2::Index, wiki_root: &Path, dir: &Path) -> Result<()> { + for entry in + std::fs::read_dir(dir).with_context(|| format!("read summary dir: {}", dir.display()))? + { + let entry = entry.with_context(|| format!("read summary dir entry: {}", dir.display()))?; + let path = entry.path(); + if path.is_dir() { + stage_summary_dir(index, wiki_root, &path)?; + } else if path.is_file() { + let repo_path = path + .strip_prefix(wiki_root) + .with_context(|| format!("summary path outside wiki root: {}", path.display()))?; + index + .add_path(repo_path) + .with_context(|| format!("stage existing wiki summary: {}", repo_path.display()))?; + } + } + Ok(()) +} + +fn commit_index_if_changed(repo: &Repository, batch: &SummaryCommitBatch) -> Result<()> { + let tree_oid = repo.index()?.write_tree()?; + let tree = repo.find_tree(tree_oid)?; + + let parent_commit = match repo.head() { + Ok(head) => Some(head.peel_to_commit()?), + Err(_) => None, + }; + + if let Some(parent) = &parent_commit { + if parent.tree_id() == tree_oid { + log::debug!( + "[content_store::wiki_git] summary wiki git clean after staging tree_id={} entries={}", + batch.tree_id, + batch.entries.len() + ); + return Ok(()); + } + } + + let sig = Signature::now(SIG_NAME, SIG_EMAIL).context("build wiki git signature")?; + let message = build_commit_message(batch); + let parents: Vec<&git2::Commit> = parent_commit.iter().collect(); + let commit_oid = repo + .commit(Some("HEAD"), &sig, &sig, &message, &tree, &parents) + .context("commit wiki summary update")?; + + log::debug!( + "[content_store::wiki_git] committed summary wiki update commit={} tree_id={} entries={}", + commit_oid, + batch.tree_id, + batch.entries.len() + ); + Ok(()) +} + +fn build_commit_message(batch: &SummaryCommitBatch) -> String { + let mut min_level = u32::MAX; + let mut max_level = 0; + let mut child_count = 0usize; + let mut token_count = 0u32; + let mut start: Option> = None; + let mut end: Option> = None; + + for entry in &batch.entries { + min_level = min_level.min(entry.level); + max_level = max_level.max(entry.level); + child_count = child_count.saturating_add(entry.child_count); + token_count = token_count.saturating_add(entry.token_count); + start = Some(start.map_or(entry.time_range_start, |s| s.min(entry.time_range_start))); + end = Some(end.map_or(entry.time_range_end, |e| e.max(entry.time_range_end))); + } + + let level_label = if min_level == max_level { + format!("L{min_level}") + } else { + format!("L{min_level}-L{max_level}") + }; + let title = format!( + "Seal memory tree {} {} summaries", + batch.tree_scope, level_label + ); + + let mut msg = String::new(); + msg.push_str(&title); + msg.push_str("\n\n"); + msg.push_str(&format!("Reason: {}\n", batch.reason)); + msg.push_str(&format!("Tree-Id: {}\n", batch.tree_id)); + msg.push_str(&format!("Tree-Scope: {}\n", batch.tree_scope)); + msg.push_str(&format!("Summary-Count: {}\n", batch.entries.len())); + msg.push_str(&format!("Level-Range: {level_label}\n")); + msg.push_str(&format!("Child-Count: {child_count}\n")); + msg.push_str(&format!("Token-Count: {token_count}\n")); + if let (Some(start), Some(end)) = (start, end) { + msg.push_str(&format!("Time-Range-Start: {}\n", start.to_rfc3339())); + msg.push_str(&format!("Time-Range-End: {}\n", end.to_rfc3339())); + } + msg.push_str("\nSummaries:\n"); + for entry in &batch.entries { + msg.push_str(&format!( + "- {} L{} children={} tokens={} path={}\n", + entry.summary_id, entry.level, entry.child_count, entry.token_count, entry.content_path + )); + } + msg +} + +fn summary_repo_path(summary_content_path: &str) -> Result { + let prefix = format!("{WIKI_PREFIX}/"); + let Some(repo_path) = summary_content_path.strip_prefix(&prefix) else { + anyhow::bail!( + "summary content path must live under {WIKI_PREFIX}/: {summary_content_path}" + ); + }; + if !repo_path.starts_with("summaries/") { + anyhow::bail!("wiki git only tracks summary nodes: {summary_content_path}"); + } + Ok(repo_path.to_string()) +} + +fn read_pointer_latest_ref(pointer_id: &str) -> String { + format!( + "refs/tags/read/{}/latest", + hex::encode(pointer_id.as_bytes()) + ) +} + +fn read_pointer_timestamp_ref(pointer_id: &str, timestamp: DateTime) -> String { + format!( + "refs/tags/read/{}/{}", + hex::encode(pointer_id.as_bytes()), + timestamp.format("%Y%m%dT%H%M%S%.9fZ") + ) +} + +#[cfg(test)] +mod tests; diff --git a/src/openhuman/memory_store/content/wiki_git/tests.rs b/src/openhuman/memory_store/content/wiki_git/tests.rs new file mode 100644 index 000000000..5814be21a --- /dev/null +++ b/src/openhuman/memory_store/content/wiki_git/tests.rs @@ -0,0 +1,328 @@ +use super::*; +use git2::IndexAddOption; +use tempfile::TempDir; + +#[test] +fn commit_summary_initializes_repo_and_tracks_only_summaries() { + let dir = TempDir::new().unwrap(); + let wiki = dir.path().join("wiki"); + let summary = wiki.join("summaries/source-slack/L1/summary-1.md"); + let raw = wiki.join("raw/should-not-track.md"); + let note = wiki.join("notes/also-ignored.md"); + std::fs::create_dir_all(summary.parent().unwrap()).unwrap(); + std::fs::create_dir_all(raw.parent().unwrap()).unwrap(); + std::fs::create_dir_all(note.parent().unwrap()).unwrap(); + std::fs::write(&summary, "---\nkind: summary\n---\nbody").unwrap(); + std::fs::write(&raw, "raw").unwrap(); + std::fs::write(¬e, "note").unwrap(); + + commit_summaries( + dir.path(), + &batch( + "queued_seal", + vec![entry( + "summary-1", + "wiki/summaries/source-slack/L1/summary-1.md", + )], + ), + ) + .unwrap(); + + let repo = Repository::open(&wiki).unwrap(); + let head = repo.head().unwrap().peel_to_commit().unwrap(); + let tree = head.tree().unwrap(); + assert!(tree.get_path(Path::new(".gitignore")).is_ok()); + assert!(tree + .get_path(Path::new("summaries/source-slack/L1/summary-1.md")) + .is_ok()); + assert!(tree.get_path(Path::new("raw/should-not-track.md")).is_err()); + assert!(tree.get_path(Path::new("notes/also-ignored.md")).is_err()); +} + +#[test] +fn commit_summary_prunes_existing_non_summary_tracked_entries() { + let dir = TempDir::new().unwrap(); + let wiki = dir.path().join("wiki"); + std::fs::create_dir_all(wiki.join("raw")).unwrap(); + std::fs::create_dir_all(wiki.join("summaries/source/L1")).unwrap(); + std::fs::write(wiki.join("raw/old.md"), "old raw").unwrap(); + std::fs::write(wiki.join("summaries/source/L1/new.md"), "new summary").unwrap(); + + let repo = Repository::init(&wiki).unwrap(); + let mut index = repo.index().unwrap(); + index + .add_all(["*"].iter(), IndexAddOption::DEFAULT, None) + .unwrap(); + index.write().unwrap(); + let tree_oid = index.write_tree().unwrap(); + let tree = repo.find_tree(tree_oid).unwrap(); + let sig = Signature::now(SIG_NAME, SIG_EMAIL).unwrap(); + repo.commit(Some("HEAD"), &sig, &sig, "old mixed commit", &tree, &[]) + .unwrap(); + + commit_summaries( + dir.path(), + &batch( + "queued_seal", + vec![entry("new", "wiki/summaries/source/L1/new.md")], + ), + ) + .unwrap(); + + let head = repo.head().unwrap().peel_to_commit().unwrap(); + let tree = head.tree().unwrap(); + assert!(tree + .get_path(Path::new("summaries/source/L1/new.md")) + .is_ok()); + assert!(tree.get_path(Path::new("raw/old.md")).is_err()); +} + +#[test] +fn commit_summary_opens_only_the_nested_wiki_repo() { + let dir = TempDir::new().unwrap(); + let wiki = dir.path().join("wiki"); + let summary = wiki.join("summaries/source/L1/summary-1.md"); + std::fs::create_dir_all(summary.parent().unwrap()).unwrap(); + std::fs::write(&summary, "summary").unwrap(); + + let parent_repo = Repository::init(dir.path()).unwrap(); + + commit_summaries( + dir.path(), + &batch( + "queued_seal", + vec![entry("summary-1", "wiki/summaries/source/L1/summary-1.md")], + ), + ) + .unwrap(); + + let repo = Repository::open(&wiki).unwrap(); + let tree = repo + .head() + .unwrap() + .peel_to_commit() + .unwrap() + .tree() + .unwrap(); + assert!(tree + .get_path(Path::new("summaries/source/L1/summary-1.md")) + .is_ok()); + assert!( + parent_repo.head().is_err(), + "summary history should not mutate the parent repo" + ); +} + +#[test] +fn commit_summary_drops_deleted_summary_entries_from_the_index() { + let dir = TempDir::new().unwrap(); + let wiki = dir.path().join("wiki"); + let old_summary = wiki.join("summaries/source/L1/old.md"); + let new_summary = wiki.join("summaries/source/L1/new.md"); + std::fs::create_dir_all(old_summary.parent().unwrap()).unwrap(); + std::fs::write(&old_summary, "old summary").unwrap(); + + commit_summaries( + dir.path(), + &batch( + "queued_seal", + vec![entry("old", "wiki/summaries/source/L1/old.md")], + ), + ) + .unwrap(); + + std::fs::remove_file(&old_summary).unwrap(); + std::fs::write(&new_summary, "new summary").unwrap(); + commit_summaries( + dir.path(), + &batch( + "queued_seal", + vec![entry("new", "wiki/summaries/source/L1/new.md")], + ), + ) + .unwrap(); + + let repo = Repository::open(&wiki).unwrap(); + let tree = repo + .head() + .unwrap() + .peel_to_commit() + .unwrap() + .tree() + .unwrap(); + assert!(tree + .get_path(Path::new("summaries/source/L1/new.md")) + .is_ok()); + assert!(tree + .get_path(Path::new("summaries/source/L1/old.md")) + .is_err()); +} + +#[test] +fn commit_summary_recovers_existing_uncommitted_summary_files() { + let dir = TempDir::new().unwrap(); + let wiki = dir.path().join("wiki"); + let missed_summary = wiki.join("summaries/source/L1/missed.md"); + let new_summary = wiki.join("summaries/source/L1/new.md"); + std::fs::create_dir_all(missed_summary.parent().unwrap()).unwrap(); + std::fs::write(&missed_summary, "missed summary").unwrap(); + std::fs::write(&new_summary, "new summary").unwrap(); + + commit_summaries( + dir.path(), + &batch( + "queued_seal", + vec![entry("new", "wiki/summaries/source/L1/new.md")], + ), + ) + .unwrap(); + + let repo = Repository::open(&wiki).unwrap(); + let tree = repo + .head() + .unwrap() + .peel_to_commit() + .unwrap() + .tree() + .unwrap(); + assert!(tree + .get_path(Path::new("summaries/source/L1/new.md")) + .is_ok()); + assert!(tree + .get_path(Path::new("summaries/source/L1/missed.md")) + .is_ok()); +} + +#[test] +fn commit_summary_rejects_non_summary_paths() { + let dir = TempDir::new().unwrap(); + let err = commit_summaries( + dir.path(), + &batch("bad", vec![entry("bad", "wiki/notes/one.md")]), + ) + .unwrap_err(); + assert!(err.to_string().contains("only tracks summary nodes")); +} + +#[test] +fn commit_message_describes_seal_metadata() { + let dir = TempDir::new().unwrap(); + let wiki = dir.path().join("wiki"); + let summary = wiki.join("summaries/source/L2/summary-2.md"); + std::fs::create_dir_all(summary.parent().unwrap()).unwrap(); + std::fs::write(&summary, "---\nkind: summary\n---\nbody").unwrap(); + + commit_summaries( + dir.path(), + &batch( + "sync_cascade", + vec![SummaryCommitEntry { + summary_id: "summary-2".to_string(), + content_path: "wiki/summaries/source/L2/summary-2.md".to_string(), + level: 2, + child_count: 7, + token_count: 123, + time_range_start: ts(1_700_000_000_000), + time_range_end: ts(1_700_003_600_000), + }], + ), + ) + .unwrap(); + + let repo = Repository::open(&wiki).unwrap(); + let head = repo.head().unwrap().peel_to_commit().unwrap(); + let msg = head.message().unwrap(); + assert!(msg.contains("Seal memory tree slack:#eng L2 summaries")); + assert!(msg.contains("Reason: sync_cascade")); + assert!(msg.contains("Summary-Count: 1")); + assert!(msg.contains("Child-Count: 7")); + assert!(msg.contains("Token-Count: 123")); + assert!(msg.contains("summary-2 L2 children=7 tokens=123")); +} + +#[test] +fn read_pointer_tags_are_timestamped_and_move_latest_without_new_commit() { + let dir = TempDir::new().unwrap(); + let wiki = dir.path().join("wiki"); + let summary = wiki.join("summaries/source/L1/summary-1.md"); + std::fs::create_dir_all(summary.parent().unwrap()).unwrap(); + std::fs::write(&summary, "---\nkind: summary\n---\nbody").unwrap(); + commit_summaries( + dir.path(), + &batch( + "queued_seal", + vec![entry("summary-1", "wiki/summaries/source/L1/summary-1.md")], + ), + ) + .unwrap(); + + let repo = Repository::open(&wiki).unwrap(); + let head = repo.head().unwrap().peel_to_commit().unwrap(); + let head_id = head.id().to_string(); + + let tagged = set_read_pointer_tag(dir.path(), "agent:default", None).unwrap(); + assert_eq!(tagged, head_id); + assert_eq!( + get_read_pointer_tag(dir.path(), "agent:default") + .unwrap() + .as_deref(), + Some(head_id.as_str()) + ); + let tag_prefix = format!( + "refs/tags/read/{}/", + hex::encode("agent:default".as_bytes()) + ); + let tags = repo.references().unwrap().fold(Vec::new(), |mut acc, r| { + let r = r.unwrap(); + let name = r.name().unwrap(); + if name.starts_with(&tag_prefix) { + acc.push(name.to_string()); + } + acc + }); + assert!( + tags.iter().any(|name| name.ends_with("/latest")), + "latest read pointer tag should be present: {tags:?}" + ); + assert!( + tags.iter().any(|name| { + let suffix = name.strip_prefix(&tag_prefix).unwrap_or_default(); + suffix.len() == "20260626T045537.123456789Z".len() + && suffix.ends_with('Z') + && suffix.contains('T') + }), + "timestamped read pointer tag should be present: {tags:?}" + ); + let mut walk = repo.revwalk().unwrap(); + walk.push_head().unwrap(); + assert_eq!( + walk.count(), + 1, + "moving the read pointer must not create commits" + ); +} + +fn batch(reason: &str, entries: Vec) -> SummaryCommitBatch { + SummaryCommitBatch { + reason: reason.to_string(), + tree_id: "tree-1".to_string(), + tree_scope: "slack:#eng".to_string(), + entries, + } +} + +fn entry(summary_id: &str, content_path: &str) -> SummaryCommitEntry { + SummaryCommitEntry { + summary_id: summary_id.to_string(), + content_path: content_path.to_string(), + level: 1, + child_count: 2, + token_count: 10, + time_range_start: ts(1_700_000_000_000), + time_range_end: ts(1_700_000_001_000), + } +} + +fn ts(ms: i64) -> DateTime { + DateTime::::from_timestamp_millis(ms).unwrap() +} diff --git a/src/openhuman/memory_tree/ingest.rs b/src/openhuman/memory_tree/ingest.rs index 6afbc335e..b5cb17733 100644 --- a/src/openhuman/memory_tree/ingest.rs +++ b/src/openhuman/memory_tree/ingest.rs @@ -14,6 +14,7 @@ use chrono::{DateTime, Utc}; use crate::openhuman::config::Config; use crate::openhuman::memory_store::chunks::store::with_connection; use crate::openhuman::memory_store::content::atomic::{stage_summary, StagedSummary}; +use crate::openhuman::memory_store::content::wiki_git::{SummaryCommitBatch, SummaryCommitEntry}; use crate::openhuman::memory_store::content::SummaryComposeInput; use crate::openhuman::memory_store::trees::types::{SummaryNode, Tree, SUMMARY_FANOUT}; use crate::openhuman::memory_tree::tree::factory::TreeFactory; @@ -137,9 +138,14 @@ pub async fn ingest_summary( version_ms: None, }; + let summary_commit = ingested_summary_batch(tree, &node, &staged.content_path); + // Persist summary + update buffer in one transaction. persist_and_buffer(config, tree, &node, &staged, target_level)?; + commit_ingested_summary(config, &summary_commit) + .with_context(|| format!("commit ingested summary {summary_id} after DB persist"))?; + // Cascade seals upward from L1 if the buffer crossed the fanout gate. let sealed_ids = cascade_from(config, tree, target_level).await?; @@ -157,6 +163,34 @@ pub async fn ingest_summary( }) } +fn ingested_summary_batch( + tree: &Tree, + node: &SummaryNode, + content_path: &str, +) -> SummaryCommitBatch { + SummaryCommitBatch { + reason: "summary_ingest".to_string(), + tree_id: tree.id.clone(), + tree_scope: tree.scope.clone(), + entries: vec![SummaryCommitEntry { + summary_id: node.id.clone(), + content_path: content_path.to_string(), + level: node.level, + child_count: node.child_ids.len(), + token_count: node.token_count, + time_range_start: node.time_range_start, + time_range_end: node.time_range_end, + }], + } +} + +fn commit_ingested_summary(config: &Config, batch: &SummaryCommitBatch) -> Result<()> { + crate::openhuman::memory_store::content::wiki_git::commit_summaries( + &config.memory_tree_content_root(), + batch, + ) +} + fn persist_and_buffer( config: &Config, tree: &Tree, diff --git a/src/openhuman/memory_tree/tree/bucket_seal.rs b/src/openhuman/memory_tree/tree/bucket_seal.rs index b78505fcd..691313ba9 100644 --- a/src/openhuman/memory_tree/tree/bucket_seal.rs +++ b/src/openhuman/memory_tree/tree/bucket_seal.rs @@ -47,10 +47,11 @@ use crate::openhuman::memory_store::chunks::store::with_connection; use crate::openhuman::memory_store::content::{ atomic::stage_summary_with_layout, paths::{slugify_source_id, SummaryDiskLayout}, + wiki_git::{SummaryCommitBatch, SummaryCommitEntry}, SummaryComposeInput, }; use crate::openhuman::memory_store::trees::types::{ - Buffer, SummaryNode, Tree, TreeKind, INPUT_TOKEN_BUDGET, OUTPUT_TOKEN_BUDGET, SUMMARY_FANOUT, + Buffer, SummaryNode, Tree, INPUT_TOKEN_BUDGET, OUTPUT_TOKEN_BUDGET, SUMMARY_FANOUT, }; use crate::openhuman::memory_tree::score::embed::build_write_embedder; use crate::openhuman::memory_tree::score::extract::EntityExtractor; @@ -714,6 +715,18 @@ pub(crate) async fn seal_one_level( node.id, staged.content_path ); + let summary_commit = summary_seal_batch( + tree, + "bucket_seal", + std::slice::from_ref(&node), + std::slice::from_ref(&staged.content_path), + ) + .with_context(|| { + format!( + "build summary git batch failed for {}; seal aborted, buffer stays unsealed for retry", + node.id + ) + })?; // Single write transaction: insert summary, clear this buffer, append // summary id to parent buffer, bump tree max_level/root if needed, @@ -724,7 +737,6 @@ pub(crate) async fn seal_one_level( let summary_id_for_closure = summary_id.clone(); let target_level_for_closure = target_level; let tree_id = tree.id.clone(); - let tree_kind = tree.kind; with_connection(config, move |conn| { let tx = conn.unchecked_transaction()?; @@ -836,6 +848,13 @@ pub(crate) async fn seal_one_level( Ok(()) })?; + commit_summary_seal(config, &summary_commit).with_context(|| { + format!( + "commit_summary_seal failed for {} after DB seal commit", + summary_id + ) + })?; + log::info!( "[tree::bucket_seal] sealed tree_id={} level={}→{} summary_id={} children={}", tree.id, @@ -876,6 +895,57 @@ fn refresh_last_sealed_tx( Ok(()) } +fn summary_seal_batch( + tree: &Tree, + reason: &str, + nodes: &[SummaryNode], + content_paths: &[String], +) -> Result { + if nodes.is_empty() { + return Ok(SummaryCommitBatch { + reason: reason.to_string(), + tree_id: tree.id.clone(), + tree_scope: tree.scope.clone(), + entries: Vec::new(), + }); + } + if nodes.len() != content_paths.len() { + anyhow::bail!( + "[tree::bucket_seal] commit_summary_seal: node/path length mismatch nodes={} paths={}", + nodes.len(), + content_paths.len() + ); + } + + let entries = nodes + .iter() + .zip(content_paths.iter()) + .map(|(node, content_path)| SummaryCommitEntry { + summary_id: node.id.clone(), + content_path: content_path.clone(), + level: node.level, + child_count: node.child_ids.len(), + token_count: node.token_count, + time_range_start: node.time_range_start, + time_range_end: node.time_range_end, + }) + .collect(); + + Ok(SummaryCommitBatch { + reason: reason.to_string(), + tree_id: tree.id.clone(), + tree_scope: tree.scope.clone(), + entries, + }) +} + +fn commit_summary_seal(config: &Config, batch: &SummaryCommitBatch) -> Result<()> { + crate::openhuman::memory_store::content::wiki_git::commit_summaries( + &config.memory_tree_content_root(), + batch, + ) +} + /// Fetch contributions for `item_ids`. At level 0 we pull from /// `mem_tree_chunks` + `mem_tree_score`; at ≥1 we pull from /// `mem_tree_summaries`. @@ -1426,6 +1496,18 @@ async fn seal_explicit_children( node.id ) })?; + let summary_commit = summary_seal_batch( + tree, + "document_subtree_seal", + std::slice::from_ref(&node), + std::slice::from_ref(&staged.content_path), + ) + .with_context(|| { + format!( + "build summary git batch failed for doc-subtree node {}; seal aborted", + node.id + ) + })?; // Persist the summary row + backlink children — NO buffer / tree-root // mutation (those belong to the merge tier). @@ -1472,6 +1554,13 @@ async fn seal_explicit_children( Ok(()) })?; + commit_summary_seal(config, &summary_commit).with_context(|| { + format!( + "commit_summary_seal failed for doc-subtree node {} after DB seal commit", + summary_id + ) + })?; + log::info!( "[tree::bucket_seal] doc-subtree sealed tree_id={} doc_id_hash={} level={}→{} summary_id={} children={}", tree.id, diff --git a/src/openhuman/memory_tree/tree/bucket_seal_tests.rs b/src/openhuman/memory_tree/tree/bucket_seal_tests.rs index bed658d32..d9440b2de 100644 --- a/src/openhuman/memory_tree/tree/bucket_seal_tests.rs +++ b/src/openhuman/memory_tree/tree/bucket_seal_tests.rs @@ -6,6 +6,7 @@ use super::*; use crate::openhuman::memory::chat::{test_override, ChatProvider, StaticChatProvider}; use crate::openhuman::memory::tree_source::registry::get_or_create_source_tree; use crate::openhuman::memory_store::content as content_store; +use crate::openhuman::memory_store::trees::types::TreeKind; use std::sync::Arc; use tempfile::TempDir; diff --git a/tests/memory_artifacts_e2e.rs b/tests/memory_artifacts_e2e.rs index 401d3458c..a0a5e456f 100644 --- a/tests/memory_artifacts_e2e.rs +++ b/tests/memory_artifacts_e2e.rs @@ -9,12 +9,17 @@ use tempfile::tempdir; use chrono::{TimeZone, Utc}; use openhuman_core::openhuman::config::Config; +use openhuman_core::openhuman::memory::tree_source::registry::get_or_create_source_tree; use openhuman_core::openhuman::memory_queue::drain_until_idle; use openhuman_core::openhuman::memory_store::content::atomic::stage_summary; use openhuman_core::openhuman::memory_store::content::obsidian::ensure_obsidian_defaults; +use openhuman_core::openhuman::memory_store::content::wiki_git::{ + get_read_pointer_tag, set_read_pointer_tag, +}; use openhuman_core::openhuman::memory_store::content::{SummaryComposeInput, SummaryTreeKind}; use openhuman_core::openhuman::memory_sync::composio::providers::slack::ingest::ingest_page_into_memory_tree; use openhuman_core::openhuman::memory_sync::composio::providers::slack::SlackMessage; +use openhuman_core::openhuman::memory_tree::ingest::{ingest_summary, SummaryIngestInput}; fn make_config(workspace_dir: &std::path::Path) -> Config { let mut config = Config::default(); @@ -108,3 +113,93 @@ async fn sync_raw_artifacts_and_mocked_summary_match_obsidian_contract() { assert!(types_body.contains("\"time_range_start\": \"date\"")); assert!(types_body.contains("\"sealed_at\": \"datetime\"")); } + +#[tokio::test] +async fn summary_ingest_records_summary_only_git_history_and_timestamped_read_tags() { + let tmp = tempdir().expect("tempdir"); + let workspace_dir = tmp.path().join("workspace"); + std::fs::create_dir_all(&workspace_dir).expect("workspace dir"); + let config = make_config(&workspace_dir); + let content_root = config.memory_tree_content_root(); + let raw_path = content_root.join("wiki/raw/not-tracked.md"); + std::fs::create_dir_all(raw_path.parent().unwrap()).expect("raw dir"); + std::fs::write(&raw_path, "should stay out of git history").expect("seed raw file"); + + let tree = get_or_create_source_tree(&config, "github:tinyhumansai/openhuman") + .expect("create source tree"); + let start = Utc.with_ymd_and_hms(2026, 6, 26, 9, 0, 0).unwrap(); + let end = Utc.with_ymd_and_hms(2026, 6, 26, 9, 30, 0).unwrap(); + let outcome = ingest_summary( + &config, + &tree, + SummaryIngestInput { + content: "Memory wiki git history now records summary-node seals.".to_string(), + token_count: 64, + entities: Vec::new(), + topics: vec!["memory".to_string()], + time_range_start: start, + time_range_end: end, + score: 0.8, + child_labels: vec!["commit:abc123".to_string(), "issue:4142".to_string()], + child_basenames: Vec::new(), + }, + ) + .await + .expect("ingest summary"); + + let wiki_root = content_root.join("wiki"); + let repo = git2::Repository::open(&wiki_root).expect("wiki git repo should be initialized"); + let head = repo.head().expect("wiki head").peel_to_commit().unwrap(); + let tree_obj = head.tree().expect("wiki commit tree"); + + let repo_summary_path = outcome + .content_path + .strip_prefix("wiki/") + .expect("summary path should live under wiki/"); + tree_obj + .get_path(std::path::Path::new(repo_summary_path)) + .expect("summary markdown should be tracked"); + assert!( + tree_obj + .get_path(std::path::Path::new("raw/not-tracked.md")) + .is_err(), + "git history should remain scoped to summaries" + ); + + let message = head.message().expect("commit message"); + assert!(message.contains("Seal memory tree github:tinyhumansai/openhuman L1 summaries")); + assert!(message.contains("Reason: summary_ingest")); + assert!(message.contains("Summary-Count: 1")); + assert!(message.contains("Child-Count: 2")); + assert!(message.contains("Token-Count: 64")); + assert!(message.contains(&outcome.summary_id)); + assert!(message.contains(repo_summary_path)); + + let head_id = head.id().to_string(); + let tagged = set_read_pointer_tag(&content_root, "agent:e2e", None) + .expect("set timestamped read pointer tag"); + assert_eq!(tagged, head_id); + assert_eq!( + get_read_pointer_tag(&content_root, "agent:e2e") + .expect("read latest pointer") + .as_deref(), + Some(head_id.as_str()) + ); + + let tag_prefix = format!("refs/tags/read/{}/", hex::encode("agent:e2e".as_bytes())); + let tags = repo.references().unwrap().fold(Vec::new(), |mut acc, r| { + let r = r.unwrap(); + let name = r.name().unwrap(); + if name.starts_with(&tag_prefix) { + acc.push(name.to_string()); + } + acc + }); + assert!(tags.iter().any(|name| name.ends_with("/latest"))); + assert!(tags.iter().any(|name| { + let suffix = name.strip_prefix(&tag_prefix).unwrap_or_default(); + suffix.len() == "20260626T090000.000000000Z".len() + && suffix.ends_with('Z') + && suffix.contains('T') + })); +}