From 48897af9b38bc81f814311066744cff8db3f83e7 Mon Sep 17 00:00:00 2001 From: sanil-23 Date: Thu, 23 Apr 2026 02:09:58 +0530 Subject: [PATCH] feat(memory): global activity digest tree (#709) (#798) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Phase 3b of the memory architecture (umbrella #711). Adds a singleton cross-source `global` tree whose purpose is time-windowed recap ("what did I do in the last 7 days?") via a time-axis-aligned hierarchy: L0 = day, L1 = week (7 dailies), L2 = month (4 weeklies), L3 = year (12 monthlies). ## What's in this PR - `global_tree/registry.rs` — singleton `get_or_create_global_tree` with the same UNIQUE-race recovery pattern Phase 3a uses for source trees (scope is the literal `"global"`). - `global_tree/digest.rs` — `end_of_day_digest(config, day, summariser)` walks every active source tree, picks one representative contribution per tree (latest L1+ intersecting the day, fallback to root), folds them with the Summariser trait into one L0 daily node, inserts it into `mem_tree_summaries`, and triggers the L0→L1→L2→L3 cascade. Idempotent on re-run (returns `DigestOutcome::Skipped` when an L0 already exists for the day). - `global_tree/seal.rs` — count-based cascade-seal with thresholds 7 (weekly), 4 (monthly), 12 (yearly). Transactional append with idempotency on (tree_id, level, item_id) to survive partial retries. - `global_tree/recap.rs` — `recap(config, window)` maps the window to a level (<2d→L0, <14d→L1, <60d→L2, else L3), fetches covering summaries, and falls back downward when the chosen level has no sealed material yet (reports `level_used` so callers can surface "best available"). Empty tree returns `None`. - `source_tree/store.rs` — adds `list_trees_by_kind` used by the digest to enumerate source trees. - `tree/mod.rs` — exports the new `global_tree` module. ## Reuses from Phase 3a - `source_tree::store` CRUD for `mem_tree_trees` / `mem_tree_summaries` / `mem_tree_buffers` (no new schema tables). - `source_tree::summariser::{Summariser, SummaryContext, SummaryInput, InertSummariser}` — honest-stub entity/topic semantics kept as-is. - `source_tree::registry::new_summary_id` for id generation. - `source_tree::types::{Tree, SummaryNode, Buffer, TreeKind::Global, TreeStatus}` — `TreeKind::Global` was already declared in Phase 3a. - Entity backfill via `tree::score::store::index_summary_entity_ids_tx` so Phase 4 retrieval can resolve "summaries mentioning X" through the same inverted index as leaves. ## Design decision: seal.rs kept as parallel impl `source_tree::bucket_seal::cascade_all_from` uses a token-budget seal policy (`TOKEN_BUDGET`), not pluggable. The global tree needs count-based thresholds per level. Rather than refactoring the stable Phase 3a seal pipeline to accept a policy (regression risk), we keep `global_tree/seal.rs` as a parallel count-based implementation that routes through the same `source_tree::store` primitives. The shape of the transaction is intentionally identical so future consolidation is straightforward when we're ready to touch the Phase 3a seal path. ## Tests 15 new tests; 176 total `memory::tree::*` passing (was 161). Coverage: - registry idempotency, ID prefix, UNIQUE-race recovery - digest empty-day, populated-day, rerun idempotency, 7-day weekly cascade - seal below-threshold, weekly-threshold, append idempotency - recap level selection across all four bands, empty-tree, L0 fallback, L1 when sealed ## Stacked on #789 Base is `feat/709-summary-trees` (Phase 3a). Merge #789 first, then rebase this branch. PR body should flag the stack order. Co-authored-by: Claude Opus 4.7 (1M context) --- .../memory/tree/global_tree/digest.rs | 535 ++++++++++++++++++ src/openhuman/memory/tree/global_tree/mod.rs | 55 ++ .../memory/tree/global_tree/recap.rs | 340 +++++++++++ .../memory/tree/global_tree/registry.rs | 138 +++++ src/openhuman/memory/tree/global_tree/seal.rs | 464 +++++++++++++++ src/openhuman/memory/tree/mod.rs | 1 + .../memory/tree/source_tree/store.rs | 20 + 7 files changed, 1553 insertions(+) create mode 100644 src/openhuman/memory/tree/global_tree/digest.rs create mode 100644 src/openhuman/memory/tree/global_tree/mod.rs create mode 100644 src/openhuman/memory/tree/global_tree/recap.rs create mode 100644 src/openhuman/memory/tree/global_tree/registry.rs create mode 100644 src/openhuman/memory/tree/global_tree/seal.rs diff --git a/src/openhuman/memory/tree/global_tree/digest.rs b/src/openhuman/memory/tree/global_tree/digest.rs new file mode 100644 index 000000000..12a5e0769 --- /dev/null +++ b/src/openhuman/memory/tree/global_tree/digest.rs @@ -0,0 +1,535 @@ +//! End-of-day digest builder for the global activity tree (#709 Phase 3b). +//! +//! Once per calendar day we walk every active source tree, collect the +//! summary material that covers that day, fold it into one cross-source +//! recap, and persist it as an L0 node in the singleton global tree. A +//! cascade then checks whether enough daily nodes have accumulated to seal +//! the weekly/monthly/yearly levels. +//! +//! Design: +//! - Populated day → exactly one L0 (daily) node emitted + cascade. +//! - Empty day (no source tree touched today) → no-op, logs the skip. +//! - The digest picks the best "representative" input from each source +//! tree in priority order: (a) the latest L1+ summary whose time range +//! intersects the target day, else (b) the most recent chunk that day's +//! L0 buffer still holds, else (c) skip that tree. This keeps the digest +//! accurate for both high-volume sources (where material has already +//! sealed into an L1) and low-volume sources (where the day's activity +//! is still in the L0 buffer). +//! - Idempotency: if an L0 daily node already exists for the target day, +//! return `DigestOutcome::Skipped` rather than emitting a duplicate. + +use anyhow::{Context, Result}; +use chrono::{DateTime, Duration, NaiveDate, TimeZone, Utc}; +use rusqlite::OptionalExtension; + +use crate::openhuman::config::Config; +use crate::openhuman::memory::tree::global_tree::registry::get_or_create_global_tree; +use crate::openhuman::memory::tree::global_tree::seal::append_daily_and_cascade; +use crate::openhuman::memory::tree::global_tree::GLOBAL_TOKEN_BUDGET; +use crate::openhuman::memory::tree::source_tree::registry::new_summary_id; +use crate::openhuman::memory::tree::source_tree::store; +use crate::openhuman::memory::tree::source_tree::summariser::{ + Summariser, SummaryContext, SummaryInput, +}; +use crate::openhuman::memory::tree::source_tree::types::{SummaryNode, Tree, TreeKind}; +use crate::openhuman::memory::tree::store::with_connection; + +/// Outcome of a single `end_of_day_digest` call — lets the caller decide +/// whether to log skip details or propagate seal counts to telemetry. +#[derive(Debug, Clone)] +pub enum DigestOutcome { + /// Emitted one L0 daily node covering `date`, and possibly cascaded + /// into higher-level seals. `sealed_ids` lists any L1/L2/L3 nodes that + /// sealed during the cascade (empty when the weekly threshold wasn't + /// crossed). + Emitted { + daily_id: String, + source_count: usize, + sealed_ids: Vec, + }, + /// No source tree had material to contribute for `date` — nothing was + /// written. + EmptyDay, + /// An L0 node already exists for `date` (e.g. this is a re-run of the + /// same day's digest). Nothing was written. + Skipped { existing_id: String }, +} + +/// Run an end-of-day digest for `day`, appending one L0 node to the global +/// tree and cascade-sealing upward if thresholds are crossed. The +/// summariser is called once to fold the per-source material into a single +/// cross-source recap. +/// +/// `day` is the calendar date in UTC the digest should cover. Callers that +/// simply want "yesterday" can pass `Utc::now().date_naive() - Duration::days(1)`. +pub async fn end_of_day_digest( + config: &Config, + day: NaiveDate, + summariser: &dyn Summariser, +) -> Result { + let (day_start, day_end) = day_bounds_utc(day)?; + log::info!( + "[global_tree::digest] end_of_day_digest day={} window=[{}, {})", + day, + day_start, + day_end + ); + + let global = get_or_create_global_tree(config)?; + + // Idempotency: check for an existing L0 daily node whose time range + // matches this day. + if let Some(existing) = find_existing_daily(config, &global.id, day_start, day_end)? { + log::info!( + "[global_tree::digest] daily already exists for {day} id={} — skipping", + existing.id + ); + return Ok(DigestOutcome::Skipped { + existing_id: existing.id, + }); + } + + // Gather one contribution per active source tree. + let source_trees = store::list_trees_by_kind(config, TreeKind::Source)?; + log::debug!( + "[global_tree::digest] scanning {} source trees", + source_trees.len() + ); + let mut inputs: Vec = Vec::with_capacity(source_trees.len()); + for source_tree in &source_trees { + match pick_source_contribution(config, source_tree, day_start, day_end)? { + Some(inp) => { + log::debug!( + "[global_tree::digest] source={} contributed id={} tokens={}", + source_tree.scope, + inp.id, + inp.token_count + ); + inputs.push(inp); + } + None => { + log::debug!( + "[global_tree::digest] source={} had no material for {day}", + source_tree.scope + ); + } + } + } + + if inputs.is_empty() { + log::info!( + "[global_tree::digest] empty day — no source trees contributed material for {day}" + ); + return Ok(DigestOutcome::EmptyDay); + } + + // Fold cross-source material into one daily recap. + let ctx = SummaryContext { + tree_id: &global.id, + tree_kind: TreeKind::Global, + target_level: 0, // daily node lives at L0 on the global tree + token_budget: GLOBAL_TOKEN_BUDGET, + }; + let output = summariser + .summarise(&inputs, &ctx) + .await + .context("summariser failed during end-of-day digest")?; + + // Envelope: time range is the day's bounds, score carries the max + // contribution score so recall still has a ranking signal. + let score = inputs + .iter() + .map(|i| i.score) + .fold(f32::NEG_INFINITY, f32::max) + .max(0.0); + + let now = Utc::now(); + let daily_id = new_summary_id(0); + let daily = SummaryNode { + id: daily_id.clone(), + tree_id: global.id.clone(), + tree_kind: TreeKind::Global, + level: 0, + parent_id: None, + child_ids: inputs.iter().map(|i| i.id.clone()).collect(), + content: output.content, + token_count: output.token_count, + entities: output.entities, + topics: output.topics, + time_range_start: day_start, + time_range_end: day_end, + score, + sealed_at: now, + deleted: false, + }; + + // Persist the daily node. Note: we do NOT backlink parent_id on the + // child summaries here — their parents are their own source trees, not + // the global tree. The global-tree child_ids are cross-source + // *references*, not ownership. + let daily_clone = daily.clone(); + let tree_id_clone = global.id.clone(); + with_connection(config, move |conn| { + let tx = conn.unchecked_transaction()?; + store::insert_summary_tx(&tx, &daily_clone)?; + // Index any entities the summariser emitted (no-op under inert). + crate::openhuman::memory::tree::score::store::index_summary_entity_ids_tx( + &tx, + &daily_clone.entities, + &daily_clone.id, + daily_clone.score, + now.timestamp_millis(), + Some(&tree_id_clone), + )?; + tx.commit()?; + Ok(()) + })?; + + log::info!( + "[global_tree::digest] emitted daily id={} sources={} tokens={}", + daily.id, + inputs.len(), + daily.token_count + ); + + // Append into L0 buffer + cascade-seal if thresholds crossed. + let sealed_ids = append_daily_and_cascade(config, &global, &daily, summariser).await?; + + Ok(DigestOutcome::Emitted { + daily_id: daily.id, + source_count: inputs.len(), + sealed_ids, + }) +} + +/// Compute [00:00, 24:00) UTC bounds for a calendar day. +fn day_bounds_utc(day: NaiveDate) -> Result<(DateTime, DateTime)> { + let start_naive = day + .and_hms_opt(0, 0, 0) + .ok_or_else(|| anyhow::anyhow!("invalid day {day} — failed to build 00:00 timestamp"))?; + let start = Utc + .from_local_datetime(&start_naive) + .single() + .ok_or_else(|| anyhow::anyhow!("non-unique UTC time for day {day}"))?; + Ok((start, start + Duration::days(1))) +} + +/// Look for an already-emitted L0 daily node for this day. Matches on +/// `tree_kind='global' AND level=0 AND time_range_start=day_start AND deleted=0`. +fn find_existing_daily( + config: &Config, + global_tree_id: &str, + day_start: DateTime, + _day_end: DateTime, +) -> Result> { + let start_ms = day_start.timestamp_millis(); + let opt_id: Option = with_connection(config, |conn| { + let id: Option = conn + .query_row( + "SELECT id FROM mem_tree_summaries + WHERE tree_id = ?1 + AND level = 0 + AND time_range_start_ms = ?2 + AND deleted = 0 + LIMIT 1", + rusqlite::params![global_tree_id, start_ms], + |r| r.get::<_, String>(0), + ) + .optional() + .context("query for existing daily node")?; + Ok(id) + })?; + match opt_id { + Some(id) => store::get_summary(config, &id), + None => Ok(None), + } +} + +/// Pick the single best contribution from one source tree for the target +/// day. Priority: +/// 1. The latest L1+ summary whose time range intersects the day. +/// 2. The tree's current root summary (any level), as a fallback when no +/// summary intersects the exact day window. +/// +/// Returns `None` when the tree has no sealed summaries at all — a +/// brand-new tree whose L0 buffer has not yet crossed the token budget. +/// Phase 3b intentionally skips such trees rather than plumbing the raw +/// L0 buffer into the digest; low-volume sources become visible once +/// either the token or time-based flush lands them in a summary. +fn pick_source_contribution( + config: &Config, + source_tree: &Tree, + day_start: DateTime, + day_end: DateTime, +) -> Result> { + let start_ms = day_start.timestamp_millis(); + let end_ms = day_end.timestamp_millis(); + let intersecting_id: Option = with_connection(config, |conn| { + let mut stmt = conn.prepare( + "SELECT id FROM mem_tree_summaries + WHERE tree_id = ?1 + AND deleted = 0 + AND time_range_start_ms < ?3 + AND time_range_end_ms >= ?2 + ORDER BY level DESC, sealed_at_ms DESC + LIMIT 1", + )?; + let row = stmt + .query_row(rusqlite::params![&source_tree.id, start_ms, end_ms], |r| { + r.get::<_, String>(0) + }) + .optional() + .context("query intersecting source summary")?; + Ok(row) + })?; + + let chosen_id = match intersecting_id { + Some(id) => Some(id), + None => source_tree.root_id.clone(), + }; + + let Some(id) = chosen_id else { + return Ok(None); + }; + + let node = match store::get_summary(config, &id)? { + Some(n) => n, + None => { + log::warn!( + "[global_tree::digest] picked id={id} for tree={} but row missing — skipping", + source_tree.scope + ); + return Ok(None); + } + }; + + Ok(Some(SummaryInput { + id: node.id, + content: format!("[{}]\n{}", source_tree.scope, node.content), + token_count: node.token_count, + entities: node.entities, + topics: node.topics, + time_range_start: node.time_range_start, + time_range_end: node.time_range_end, + score: node.score, + })) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::openhuman::memory::tree::source_tree::bucket_seal::{append_leaf, LeafRef}; + use crate::openhuman::memory::tree::source_tree::registry::get_or_create_source_tree; + use crate::openhuman::memory::tree::source_tree::summariser::inert::InertSummariser; + use crate::openhuman::memory::tree::source_tree::types::TreeStatus; + use crate::openhuman::memory::tree::store::upsert_chunks; + use crate::openhuman::memory::tree::types::{chunk_id, Chunk, Metadata, SourceKind, SourceRef}; + use tempfile::TempDir; + + fn test_config() -> (TempDir, Config) { + let tmp = TempDir::new().unwrap(); + let mut cfg = Config::default(); + cfg.workspace_dir = tmp.path().to_path_buf(); + (tmp, cfg) + } + + async fn seed_source_tree_with_sealed_l1(cfg: &Config, scope: &str, ts: DateTime) { + // Use chunks with the source_tree bucket-seal mechanics so we get a + // real L1 summary persisted that intersects the target day. + let tree = get_or_create_source_tree(cfg, scope).unwrap(); + let summariser = InertSummariser::new(); + + let c1 = Chunk { + id: chunk_id(SourceKind::Chat, scope, 0), + content: format!("chunk 1 in {scope}"), + metadata: Metadata { + source_kind: SourceKind::Chat, + source_id: scope.into(), + owner: "alice".into(), + timestamp: ts, + time_range: (ts, ts), + tags: vec![], + source_ref: Some(SourceRef::new("slack://x")), + }, + token_count: 6_000, + seq_in_source: 0, + created_at: ts, + }; + let c2 = Chunk { + id: chunk_id(SourceKind::Chat, scope, 1), + content: format!("chunk 2 in {scope}"), + metadata: Metadata { + source_kind: SourceKind::Chat, + source_id: scope.into(), + owner: "alice".into(), + timestamp: ts, + time_range: (ts, ts), + tags: vec![], + source_ref: Some(SourceRef::new("slack://y")), + }, + token_count: 6_000, + seq_in_source: 1, + created_at: ts, + }; + upsert_chunks(cfg, &[c1.clone(), c2.clone()]).unwrap(); + + let leaf1 = LeafRef { + chunk_id: c1.id.clone(), + token_count: 6_000, + timestamp: ts, + content: c1.content.clone(), + entities: vec![], + topics: vec![], + score: 0.5, + }; + let leaf2 = LeafRef { + chunk_id: c2.id.clone(), + token_count: 6_000, + timestamp: ts, + content: c2.content.clone(), + entities: vec![], + topics: vec![], + score: 0.5, + }; + append_leaf(cfg, &tree, &leaf1, &summariser).await.unwrap(); + append_leaf(cfg, &tree, &leaf2, &summariser).await.unwrap(); + // 12k tokens > 10k budget → one L1 summary covering `ts`. + } + + #[tokio::test] + async fn empty_day_returns_empty_day_outcome() { + // No source trees exist yet — digest should no-op. + let (_tmp, cfg) = test_config(); + let summariser = InertSummariser::new(); + let day = NaiveDate::from_ymd_opt(2025, 1, 15).unwrap(); + let outcome = end_of_day_digest(&cfg, day, &summariser).await.unwrap(); + assert!(matches!(outcome, DigestOutcome::EmptyDay)); + + // No L0 nodes emitted on the global tree. + let global = get_or_create_global_tree(&cfg).unwrap(); + assert_eq!(store::count_summaries(&cfg, &global.id).unwrap(), 0); + } + + #[tokio::test] + async fn populated_day_emits_one_daily_leaf() { + let (_tmp, cfg) = test_config(); + let summariser = InertSummariser::new(); + + // Seed 3 source trees with sealed L1s on the target day. This + // exercises the main cross-source path end to end. + let day = NaiveDate::from_ymd_opt(2025, 1, 15).unwrap(); + let ts = day.and_hms_opt(12, 0, 0).unwrap().and_utc(); + seed_source_tree_with_sealed_l1(&cfg, "slack:#eng", ts).await; + seed_source_tree_with_sealed_l1(&cfg, "email:alice", ts).await; + seed_source_tree_with_sealed_l1(&cfg, "notion:workspace", ts).await; + + let outcome = end_of_day_digest(&cfg, day, &summariser).await.unwrap(); + let (daily_id, source_count) = match outcome { + DigestOutcome::Emitted { + daily_id, + source_count, + sealed_ids, + } => { + assert!(sealed_ids.is_empty(), "one day ≠ weekly seal yet"); + (daily_id, source_count) + } + other => panic!("expected Emitted, got {other:?}"), + }; + assert_eq!(source_count, 3); + + let global = get_or_create_global_tree(&cfg).unwrap(); + // Exactly one L0 daily node on the global tree. + let daily_nodes = store::list_summaries_at_level(&cfg, &global.id, 0).unwrap(); + assert_eq!(daily_nodes.len(), 1); + assert_eq!(daily_nodes[0].id, daily_id); + assert_eq!(daily_nodes[0].tree_kind, TreeKind::Global); + + // Time range matches the target day exactly. + let (expected_start, expected_end) = day_bounds_utc(day).unwrap(); + assert_eq!(daily_nodes[0].time_range_start, expected_start); + assert_eq!(daily_nodes[0].time_range_end, expected_end); + assert_eq!(daily_nodes[0].child_ids.len(), 3); + + // L0 buffer now carries this daily id (≠ empty). + let l0 = store::get_buffer(&cfg, &global.id, 0).unwrap(); + assert_eq!(l0.item_ids, vec![daily_id]); + } + + #[tokio::test] + async fn rerun_on_same_day_is_idempotent() { + let (_tmp, cfg) = test_config(); + let summariser = InertSummariser::new(); + let day = NaiveDate::from_ymd_opt(2025, 2, 3).unwrap(); + let ts = day.and_hms_opt(9, 0, 0).unwrap().and_utc(); + seed_source_tree_with_sealed_l1(&cfg, "slack:#eng", ts).await; + + let first = end_of_day_digest(&cfg, day, &summariser).await.unwrap(); + let first_id = match first { + DigestOutcome::Emitted { daily_id, .. } => daily_id, + other => panic!("expected Emitted, got {other:?}"), + }; + + let second = end_of_day_digest(&cfg, day, &summariser).await.unwrap(); + match second { + DigestOutcome::Skipped { existing_id } => assert_eq!(existing_id, first_id), + other => panic!("expected Skipped on rerun, got {other:?}"), + } + + let global = get_or_create_global_tree(&cfg).unwrap(); + let daily_nodes = store::list_summaries_at_level(&cfg, &global.id, 0).unwrap(); + assert_eq!(daily_nodes.len(), 1, "rerun must not duplicate daily node"); + } + + #[tokio::test] + async fn seven_days_cascade_to_weekly_seal() { + let (_tmp, cfg) = test_config(); + let summariser = InertSummariser::new(); + + // Emit 7 daily nodes across 7 consecutive days. The 7th should + // cascade to seal an L1 weekly node. + let base = NaiveDate::from_ymd_opt(2025, 3, 1).unwrap(); + let mut emitted_days = 0; + for i in 0..7 { + let day = base + Duration::days(i); + let ts = day.and_hms_opt(10, 0, 0).unwrap().and_utc(); + // Fresh source scope per day keeps L1s day-specific. + seed_source_tree_with_sealed_l1(&cfg, &format!("slack:#day{i}"), ts).await; + + let outcome = end_of_day_digest(&cfg, day, &summariser).await.unwrap(); + match outcome { + DigestOutcome::Emitted { + sealed_ids, + source_count: _, + .. + } => { + emitted_days += 1; + if emitted_days < 7 { + assert!( + sealed_ids.is_empty(), + "no weekly seal until 7 daily nodes accumulate" + ); + } else { + assert_eq!(sealed_ids.len(), 1, "weekly seal should fire on day 7"); + } + } + other => panic!("expected Emitted on day {i}, got {other:?}"), + } + } + assert_eq!(emitted_days, 7); + + let global = get_or_create_global_tree(&cfg).unwrap(); + let l0 = store::get_buffer(&cfg, &global.id, 0).unwrap(); + assert!(l0.is_empty(), "L0 buffer cleared after weekly seal"); + let l1 = store::get_buffer(&cfg, &global.id, 1).unwrap(); + assert_eq!(l1.item_ids.len(), 1, "one weekly node parked at L1"); + + let weekly = store::get_summary(&cfg, &l1.item_ids[0]).unwrap().unwrap(); + assert_eq!(weekly.level, 1); + assert_eq!(weekly.child_ids.len(), 7); + + let t = store::get_tree(&cfg, &global.id).unwrap().unwrap(); + assert_eq!(t.max_level, 1); + assert_eq!(t.status, TreeStatus::Active); + } +} diff --git a/src/openhuman/memory/tree/global_tree/mod.rs b/src/openhuman/memory/tree/global_tree/mod.rs new file mode 100644 index 000000000..30ce33df3 --- /dev/null +++ b/src/openhuman/memory/tree/global_tree/mod.rs @@ -0,0 +1,55 @@ +//! Phase 3b — Global Activity Digest tree (#709 umbrella, spec in +//! `docs/MEMORY_ARCHITECTURE_LLD.md`). +//! +//! The global tree is a **single cross-source recap structure**: one tree +//! per workspace, built end-of-day from the source trees' current roots so +//! a later question like "what did I do in the last 7 days?" can be +//! answered with one summary hop. Unlike source trees whose L0 holds raw +//! chunk leaves, the global tree's L0 already holds synthesised daily +//! summaries — each one a fold of the day's activity across every active +//! source tree. +//! +//! Level conventions (time-axis aligned, not token-driven): +//! - L0 = one node per **day** (emitted by `end_of_day_digest`) +//! - L1 = one node per **week** (~7 daily leaves) +//! - L2 = one node per **month** (~4 weekly nodes) +//! - L3 = one node per **year** (~12 monthly nodes) +//! +//! Reuses Phase 3a storage (`mem_tree_trees`, `mem_tree_summaries`, +//! `mem_tree_buffers` with `kind='global'`) and the `Summariser` trait. +//! The `InertSummariser` fallback is explicitly an honest stub — entities +//! and topics stay empty until an LLM-backed summariser lands. +//! +//! Public surface at Phase 3b: +//! - [`registry::get_or_create_global_tree`] — singleton (scope="global") +//! - [`digest::end_of_day_digest`] — build one L0 daily node, cascade-seal +//! - [`recap::recap`] — select the right level for a time window + +pub mod digest; +pub mod recap; +pub mod registry; +pub mod seal; + +pub use digest::{end_of_day_digest, DigestOutcome}; +pub use recap::{recap, RecapOutput}; +pub use registry::get_or_create_global_tree; + +/// Number of L0 (daily) nodes that seal into one L1 (weekly) node. +pub const WEEKLY_SEAL_THRESHOLD: usize = 7; + +/// Number of L1 (weekly) nodes that seal into one L2 (monthly) node. +/// ~4.35 weeks per month; we round down to seal monthly-ish when enough +/// weekly material accumulates. +pub const MONTHLY_SEAL_THRESHOLD: usize = 4; + +/// Number of L2 (monthly) nodes that seal into one L3 (yearly) node. +pub const YEARLY_SEAL_THRESHOLD: usize = 12; + +/// Literal scope used for the singleton global tree. +pub const GLOBAL_SCOPE: &str = "global"; + +/// Token budget passed into the summariser for global-tree seals. The +/// token-based seal trigger is disabled on the global tree (we use a +/// time/count trigger instead), so this is purely a ceiling on the +/// summariser's output length at each level. +pub const GLOBAL_TOKEN_BUDGET: u32 = 4_000; diff --git a/src/openhuman/memory/tree/global_tree/recap.rs b/src/openhuman/memory/tree/global_tree/recap.rs new file mode 100644 index 000000000..c99e3b271 --- /dev/null +++ b/src/openhuman/memory/tree/global_tree/recap.rs @@ -0,0 +1,340 @@ +//! Window-scoped recap retrieval for the global activity tree (#709 Phase 3b). +//! +//! Given a duration (e.g. `Duration::days(7)`), pick the tree level that +//! naturally matches the time axis and return the latest summary at that +//! level. This is the read half of the global digest: the digest builder +//! plants daily/weekly/monthly/yearly nodes, and `recap` retrieves the one +//! best suited for the caller's question. +//! +//! Level selection (width thresholds chosen to cover expected call sites): +//! - `< 2 days` → latest L0 (today's digest) +//! - `< 14 days` → latest L1 (weekly) +//! - `< 60 days` → latest L2 (monthly) +//! - `≥ 60 days` → latest L3 (yearly), padded with the covering L2s when no L3 has sealed yet. +//! +//! When no summary exists at the chosen level, the function falls back +//! downward (to the latest lower-level node) and reports the actual level +//! used in the `level_used` field of the result so callers can surface +//! "best available" to users. Returns `None` only when the global tree has +//! no sealed summaries at all. + +use anyhow::Result; +use chrono::{DateTime, Duration, Utc}; + +use crate::openhuman::config::Config; +use crate::openhuman::memory::tree::global_tree::registry::get_or_create_global_tree; +use crate::openhuman::memory::tree::source_tree::store; +use crate::openhuman::memory::tree::source_tree::types::SummaryNode; + +/// Aggregated recap returned to the caller. +#[derive(Debug, Clone)] +pub struct RecapOutput { + /// The rolled-up content for the chosen window. + pub content: String, + /// The time span actually covered by the returned content. Start is the + /// earliest `time_range_start` across included summaries, end is the + /// latest `time_range_end`. + pub time_range: (DateTime, DateTime), + /// The level actually used to build the recap. May be lower than the + /// requested level when the higher level has no sealed nodes yet. + pub level_used: u32, + /// One entry per summary folded into the content, in the order they + /// were concatenated. Lets callers surface provenance ("this recap + /// covers weekly summaries W, W-1, W-2"). + pub summary_ids: Vec, +} + +/// Return a recap for the given window, or `None` if no global summaries +/// have sealed yet. +pub async fn recap(config: &Config, window: Duration) -> Result> { + let target_level = pick_level(window); + log::info!( + "[global_tree::recap] recap window={:?} target_level={}", + window, + target_level + ); + + let global = get_or_create_global_tree(config)?; + let now = Utc::now(); + let window_start = now - window; + + // Walk down from `target_level` to 0 looking for material. + for level in (0..=target_level).rev() { + let all_at_level = store::list_summaries_at_level(config, &global.id, level)?; + if all_at_level.is_empty() { + continue; + } + let covering = pick_covering(&all_at_level, window_start, now); + if covering.is_empty() { + continue; + } + log::debug!( + "[global_tree::recap] using level={} summaries={}", + level, + covering.len() + ); + return Ok(Some(assemble_recap(&covering, level))); + } + + log::info!("[global_tree::recap] no global summaries yet — nothing to recap"); + Ok(None) +} + +/// Map a window duration to the level whose node-granularity best matches +/// the window. See module-level doc for the thresholds. +pub fn pick_level(window: Duration) -> u32 { + // Direct comparisons keep the selection readable versus a table walk + // since there are only four bands. See module-level doc for the exact + // ceilings. + if window < Duration::days(2) { + 0 + } else if window < Duration::days(14) { + 1 + } else if window < Duration::days(60) { + 2 + } else { + 3 + } +} + +/// Select every summary at the given level whose time range overlaps the +/// [window_start, now] window, ordered oldest → newest. When none overlap +/// (a long quiet stretch ending before the window) we fall back to the +/// single latest summary so callers still get *something* useful. +fn pick_covering( + summaries: &[SummaryNode], + window_start: DateTime, + now: DateTime, +) -> Vec<&SummaryNode> { + let mut overlapping: Vec<&SummaryNode> = summaries + .iter() + .filter(|s| s.time_range_end >= window_start && s.time_range_start <= now) + .collect(); + overlapping.sort_by_key(|s| s.time_range_start); + + if overlapping.is_empty() { + if let Some(latest) = summaries.iter().max_by_key(|s| s.sealed_at) { + return vec![latest]; + } + } + overlapping +} + +/// Concatenate the selected summaries with provenance markers and compute +/// the time envelope. +fn assemble_recap(covering: &[&SummaryNode], level: u32) -> RecapOutput { + let mut parts: Vec = Vec::with_capacity(covering.len()); + let mut summary_ids: Vec = Vec::with_capacity(covering.len()); + for s in covering { + parts.push(format!( + "[{} → {}]\n{}", + s.time_range_start.to_rfc3339(), + s.time_range_end.to_rfc3339(), + s.content + )); + summary_ids.push(s.id.clone()); + } + let content = parts.join("\n\n"); + + let time_start = covering + .iter() + .map(|s| s.time_range_start) + .min() + .unwrap_or_else(Utc::now); + let time_end = covering + .iter() + .map(|s| s.time_range_end) + .max() + .unwrap_or_else(Utc::now); + + RecapOutput { + content, + time_range: (time_start, time_end), + level_used: level, + summary_ids, + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::openhuman::memory::tree::global_tree::digest::{end_of_day_digest, DigestOutcome}; + use crate::openhuman::memory::tree::source_tree::bucket_seal::{append_leaf, LeafRef}; + use crate::openhuman::memory::tree::source_tree::registry::get_or_create_source_tree; + use crate::openhuman::memory::tree::source_tree::summariser::inert::InertSummariser; + use crate::openhuman::memory::tree::store::upsert_chunks; + use crate::openhuman::memory::tree::types::{chunk_id, Chunk, Metadata, SourceKind, SourceRef}; + use tempfile::TempDir; + + fn test_config() -> (TempDir, Config) { + let tmp = TempDir::new().unwrap(); + let mut cfg = Config::default(); + cfg.workspace_dir = tmp.path().to_path_buf(); + (tmp, cfg) + } + + #[test] + fn pick_level_matches_window_thresholds() { + assert_eq!(pick_level(Duration::hours(1)), 0); + assert_eq!(pick_level(Duration::days(1)), 0); + assert_eq!(pick_level(Duration::days(2)), 1); + assert_eq!(pick_level(Duration::days(7)), 1); + assert_eq!(pick_level(Duration::days(13)), 1); + assert_eq!(pick_level(Duration::days(14)), 2); + assert_eq!(pick_level(Duration::days(30)), 2); + assert_eq!(pick_level(Duration::days(59)), 2); + assert_eq!(pick_level(Duration::days(60)), 3); + assert_eq!(pick_level(Duration::days(365)), 3); + } + + #[tokio::test] + async fn recap_on_empty_tree_returns_none() { + let (_tmp, cfg) = test_config(); + let out = recap(&cfg, Duration::days(7)).await.unwrap(); + assert!(out.is_none()); + } + + async fn seed_source_l1(cfg: &Config, scope: &str, ts: DateTime) { + let tree = get_or_create_source_tree(cfg, scope).unwrap(); + let summariser = InertSummariser::new(); + let c1 = Chunk { + id: chunk_id(SourceKind::Chat, scope, 0), + content: format!("c1-{scope}"), + metadata: Metadata { + source_kind: SourceKind::Chat, + source_id: scope.into(), + owner: "alice".into(), + timestamp: ts, + time_range: (ts, ts), + tags: vec![], + source_ref: Some(SourceRef::new("slack://x")), + }, + token_count: 6_000, + seq_in_source: 0, + created_at: ts, + }; + let c2 = Chunk { + id: chunk_id(SourceKind::Chat, scope, 1), + content: format!("c2-{scope}"), + metadata: Metadata { + source_kind: SourceKind::Chat, + source_id: scope.into(), + owner: "alice".into(), + timestamp: ts, + time_range: (ts, ts), + tags: vec![], + source_ref: Some(SourceRef::new("slack://y")), + }, + token_count: 6_000, + seq_in_source: 1, + created_at: ts, + }; + upsert_chunks(cfg, &[c1.clone(), c2.clone()]).unwrap(); + append_leaf( + cfg, + &tree, + &LeafRef { + chunk_id: c1.id.clone(), + token_count: 6_000, + timestamp: ts, + content: c1.content.clone(), + entities: vec![], + topics: vec![], + score: 0.5, + }, + &summariser, + ) + .await + .unwrap(); + append_leaf( + cfg, + &tree, + &LeafRef { + chunk_id: c2.id.clone(), + token_count: 6_000, + timestamp: ts, + content: c2.content.clone(), + entities: vec![], + topics: vec![], + score: 0.5, + }, + &summariser, + ) + .await + .unwrap(); + } + + #[tokio::test] + async fn recap_one_day_window_returns_latest_l0() { + // One daily digest → recap(1 day) should return the L0 at the + // correct level. + let (_tmp, cfg) = test_config(); + let summariser = InertSummariser::new(); + // Use "today" so the digest's time range covers now. + let day = Utc::now().date_naive(); + let ts = day.and_hms_opt(12, 0, 0).unwrap().and_utc(); + seed_source_l1(&cfg, "slack:#eng", ts).await; + let outcome = end_of_day_digest(&cfg, day, &summariser).await.unwrap(); + assert!(matches!(outcome, DigestOutcome::Emitted { .. })); + + let r = recap(&cfg, Duration::hours(24)) + .await + .unwrap() + .expect("expected a recap with one daily node emitted"); + assert_eq!(r.level_used, 0); + assert_eq!(r.summary_ids.len(), 1); + assert!(!r.content.is_empty()); + } + + #[tokio::test] + async fn recap_weekly_window_falls_back_to_l0_when_no_l1() { + // With only 3 daily nodes (< 7) no L1 has sealed. A 7-day recap + // should fall back from level 1 to level 0 and return whatever + // daily nodes exist. + let (_tmp, cfg) = test_config(); + let summariser = InertSummariser::new(); + let today = Utc::now().date_naive(); + let base = today - Duration::days(2); + for i in 0..3 { + let day = base + Duration::days(i); + let ts = day.and_hms_opt(10, 0, 0).unwrap().and_utc(); + seed_source_l1(&cfg, &format!("slack:#d{i}"), ts).await; + end_of_day_digest(&cfg, day, &summariser).await.unwrap(); + } + let r = recap(&cfg, Duration::days(7)) + .await + .unwrap() + .expect("expected fallback recap"); + assert_eq!( + r.level_used, 0, + "no L1 available yet → recap falls back to L0" + ); + assert_eq!(r.summary_ids.len(), 3, "all three daily nodes folded in"); + } + + #[tokio::test] + async fn recap_weekly_window_uses_l1_when_sealed() { + // After 7 daily digests a weekly L1 exists. A 7-day recap should + // return that L1 at level 1. + let (_tmp, cfg) = test_config(); + let summariser = InertSummariser::new(); + let today = Utc::now().date_naive(); + let base = today - Duration::days(6); + for i in 0..7 { + let day = base + Duration::days(i); + let ts = day.and_hms_opt(10, 0, 0).unwrap().and_utc(); + seed_source_l1(&cfg, &format!("slack:#w{i}"), ts).await; + end_of_day_digest(&cfg, day, &summariser).await.unwrap(); + } + let r = recap(&cfg, Duration::days(7)) + .await + .unwrap() + .expect("expected recap with weekly seal"); + assert_eq!(r.level_used, 1); + assert_eq!( + r.summary_ids.len(), + 1, + "one weekly L1 node covers the window" + ); + } +} diff --git a/src/openhuman/memory/tree/global_tree/registry.rs b/src/openhuman/memory/tree/global_tree/registry.rs new file mode 100644 index 000000000..c71452602 --- /dev/null +++ b/src/openhuman/memory/tree/global_tree/registry.rs @@ -0,0 +1,138 @@ +//! Singleton registry for the global activity digest tree (#709, Phase 3b). +//! +//! Unlike source trees (one per `source_id`) the global tree is a true +//! singleton per workspace — scope is the literal string `"global"`. The +//! lookup and race-recovery pattern otherwise mirrors +//! `source_tree::registry::get_or_create_source_tree`. + +use anyhow::Result; +use chrono::Utc; +use uuid::Uuid; + +use crate::openhuman::config::Config; +use crate::openhuman::memory::tree::global_tree::GLOBAL_SCOPE; +use crate::openhuman::memory::tree::source_tree::store; +use crate::openhuman::memory::tree::source_tree::types::{Tree, TreeKind, TreeStatus}; + +/// Return the workspace's singleton global tree, creating it lazily on +/// first call. Safe to call on every ingest; subsequent calls short-circuit +/// to the existing row. +pub fn get_or_create_global_tree(config: &Config) -> Result { + if let Some(existing) = store::get_tree_by_scope(config, TreeKind::Global, GLOBAL_SCOPE)? { + log::debug!( + "[global_tree::registry] found global tree id={}", + existing.id + ); + return Ok(existing); + } + + let tree = Tree { + id: new_global_tree_id(), + kind: TreeKind::Global, + scope: GLOBAL_SCOPE.to_string(), + root_id: None, + max_level: 0, + status: TreeStatus::Active, + created_at: Utc::now(), + last_sealed_at: None, + }; + match store::insert_tree(config, &tree) { + Ok(()) => { + log::info!("[global_tree::registry] created global tree id={}", tree.id); + Ok(tree) + } + Err(err) if is_unique_violation(&err) => { + // Another caller beat us to it between our initial lookup and + // the insert. The UNIQUE(kind, scope) index caught it — + // re-query and return the winner. + log::debug!("[global_tree::registry] UNIQUE race for global tree — re-querying"); + store::get_tree_by_scope(config, TreeKind::Global, GLOBAL_SCOPE)?.ok_or_else(|| { + anyhow::anyhow!( + "UNIQUE violation on global-tree insert but no row found on re-query" + ) + }) + } + Err(err) => Err(err), + } +} + +/// True when `err` wraps a SQLite UNIQUE constraint violation. Duplicated +/// from `source_tree::registry` to keep this module self-contained; the +/// two copies are ~5 lines and have the same shape. +fn is_unique_violation(err: &anyhow::Error) -> bool { + if let Some(rusqlite::Error::SqliteFailure(sqlite_err, _)) = + err.downcast_ref::() + { + return sqlite_err.code == rusqlite::ErrorCode::ConstraintViolation; + } + let msg = format!("{err:#}"); + msg.contains("UNIQUE constraint failed") +} + +fn new_global_tree_id() -> String { + format!("{}:{}", TreeKind::Global.as_str(), Uuid::new_v4()) +} + +#[cfg(test)] +mod tests { + use super::*; + use tempfile::TempDir; + + fn test_config() -> (TempDir, Config) { + let tmp = TempDir::new().unwrap(); + let mut cfg = Config::default(); + cfg.workspace_dir = tmp.path().to_path_buf(); + (tmp, cfg) + } + + #[test] + fn get_or_create_is_idempotent() { + let (_tmp, cfg) = test_config(); + let first = get_or_create_global_tree(&cfg).unwrap(); + let second = get_or_create_global_tree(&cfg).unwrap(); + assert_eq!(first.id, second.id); + assert_eq!(first.kind, TreeKind::Global); + assert_eq!(first.scope, GLOBAL_SCOPE); + assert_eq!(first.status, TreeStatus::Active); + } + + #[test] + fn global_tree_has_expected_id_prefix() { + let id = new_global_tree_id(); + assert!(id.starts_with("global:")); + } + + #[test] + fn race_recovery_returns_existing_row() { + // Pre-seed a global tree so the second `get_or_create` path exercises + // the normal lookup branch; the UNIQUE-race branch is covered by the + // shared `is_unique_violation` contract in `source_tree::registry`. + let (_tmp, cfg) = test_config(); + let pre_existing = Tree { + id: "global:preexisting".into(), + kind: TreeKind::Global, + scope: GLOBAL_SCOPE.into(), + root_id: None, + max_level: 0, + status: TreeStatus::Active, + created_at: Utc::now(), + last_sealed_at: None, + }; + store::insert_tree(&cfg, &pre_existing).unwrap(); + + let got = get_or_create_global_tree(&cfg).unwrap(); + assert_eq!(got.id, "global:preexisting"); + + // And a direct duplicate insert must fire UNIQUE, covering the + // detector path this module depends on for race recovery. + let dup = Tree { + id: "global:would-collide".into(), + ..pre_existing.clone() + }; + let err = store::insert_tree(&cfg, &dup).unwrap_err(); + assert!( + is_unique_violation(&err), + "expected UNIQUE violation, got {err:#}" + ); + } +} diff --git a/src/openhuman/memory/tree/global_tree/seal.rs b/src/openhuman/memory/tree/global_tree/seal.rs new file mode 100644 index 000000000..8f3c787d5 --- /dev/null +++ b/src/openhuman/memory/tree/global_tree/seal.rs @@ -0,0 +1,464 @@ +//! Count-based cascade-seal for the global activity digest tree (#709 Phase 3b). +//! +//! The global tree's trigger is **time/count-based**, not token-based: seal +//! L0 → L1 when 7 daily nodes accumulate, L1 → L2 when 4 weekly nodes +//! accumulate, L2 → L3 when 12 monthly nodes accumulate. This keeps the +//! tree aligned to the time axis (day / week / month / year) so +//! window-scoped recap queries can map a duration to a level deterministically. +//! +//! Reuses Phase 3a storage primitives from `source_tree::store` without +//! their token-budget cascade logic — all global seals route through +//! `mem_tree_summaries` on both sides (children and output), since even L0 +//! is a sealed summary node rather than a raw chunk. + +use anyhow::{Context, Result}; +use chrono::{DateTime, Utc}; + +use crate::openhuman::config::Config; +use crate::openhuman::memory::tree::global_tree::{ + GLOBAL_TOKEN_BUDGET, MONTHLY_SEAL_THRESHOLD, WEEKLY_SEAL_THRESHOLD, YEARLY_SEAL_THRESHOLD, +}; +use crate::openhuman::memory::tree::source_tree::registry::new_summary_id; +use crate::openhuman::memory::tree::source_tree::store; +use crate::openhuman::memory::tree::source_tree::summariser::{ + Summariser, SummaryContext, SummaryInput, +}; +use crate::openhuman::memory::tree::source_tree::types::{Buffer, SummaryNode, Tree, TreeKind}; +use crate::openhuman::memory::tree::store::with_connection; + +/// Hard cap on cascade depth — mirrors the source-tree constant. L0→L1→L2→L3 +/// is only 3 hops so we have ample slack. +const MAX_CASCADE_DEPTH: u32 = 32; + +/// Idempotently append one level-0 (daily) summary id to the global tree's +/// L0 buffer, then cascade-seal upward if count thresholds are crossed. +/// +/// The caller (`digest::end_of_day_digest`) has already inserted the L0 +/// node into `mem_tree_summaries`; this function only handles the buffer +/// accounting + cascade. +pub async fn append_daily_and_cascade( + config: &Config, + tree: &Tree, + daily_summary: &SummaryNode, + summariser: &dyn Summariser, +) -> Result> { + log::debug!( + "[global_tree::seal] append_daily tree_id={} daily_id={} tokens={}", + tree.id, + daily_summary.id, + daily_summary.token_count + ); + + append_to_buffer( + config, + &tree.id, + 0, + &daily_summary.id, + daily_summary.token_count as i64, + daily_summary.time_range_start, + )?; + + cascade_seals(config, tree, summariser).await +} + +/// Transactionally append a single summary id to the buffer at +/// `(tree_id, level)`. Idempotent on the `(tree_id, level, item_id)` tuple +/// so retries of a partially-applied digest don't double-count. +fn append_to_buffer( + config: &Config, + tree_id: &str, + level: u32, + item_id: &str, + token_delta: i64, + item_ts: DateTime, +) -> Result<()> { + with_connection(config, |conn| { + let tx = conn.unchecked_transaction()?; + let mut buf = store::get_buffer_conn(&tx, tree_id, level)?; + if buf.item_ids.iter().any(|existing| existing == item_id) { + log::debug!( + "[global_tree::seal] append_to_buffer: {item_id} already in buffer \ + tree_id={tree_id} level={level} — no-op" + ); + return Ok(()); + } + buf.item_ids.push(item_id.to_string()); + buf.token_sum = buf.token_sum.saturating_add(token_delta); + buf.oldest_at = match buf.oldest_at { + Some(existing) => Some(existing.min(item_ts)), + None => Some(item_ts), + }; + store::upsert_buffer_tx(&tx, &buf)?; + tx.commit()?; + Ok(()) + }) +} + +async fn cascade_seals( + config: &Config, + tree: &Tree, + summariser: &dyn Summariser, +) -> Result> { + let mut sealed_ids: Vec = Vec::new(); + // `level` is independent of the iteration counter — it only bumps when a + // seal fires, and the loop can break early if `should_seal` returns + // false. Clippy's loop-counter suggestion would merge them incorrectly. + #[allow(clippy::explicit_counter_loop)] + { + let mut level: u32 = 0; + for _ in 0..MAX_CASCADE_DEPTH { + let buf = store::get_buffer(config, &tree.id, level)?; + if !should_seal(&buf, level) { + log::debug!( + "[global_tree::seal] cascade done tree_id={} stop_level={} count={}", + tree.id, + level, + buf.item_ids.len() + ); + break; + } + + let summary_id = seal_one_level(config, tree, &buf, summariser).await?; + sealed_ids.push(summary_id); + level += 1; + } + } + + Ok(sealed_ids) +} + +/// Count-based threshold per level. L0→L1 needs 7 daily nodes, L1→L2 needs +/// 4 weekly nodes, L2→L3 needs 12 monthly nodes. Levels ≥ 3 never seal in +/// this phase — a yearly node is the top of the global tree. +fn should_seal(buf: &Buffer, level: u32) -> bool { + let threshold = match level { + 0 => WEEKLY_SEAL_THRESHOLD, + 1 => MONTHLY_SEAL_THRESHOLD, + 2 => YEARLY_SEAL_THRESHOLD, + _ => return false, + }; + !buf.is_empty() && buf.item_ids.len() >= threshold +} + +async fn seal_one_level( + config: &Config, + tree: &Tree, + buf: &Buffer, + summariser: &dyn Summariser, +) -> Result { + let level = buf.level; + let target_level = level + 1; + + let inputs = hydrate_summary_inputs(config, &buf.item_ids)?; + if inputs.is_empty() { + anyhow::bail!( + "[global_tree::seal] refused to seal empty buffer tree_id={} level={}", + tree.id, + level + ); + } + + let time_range_start = inputs + .iter() + .map(|i| i.time_range_start) + .min() + .unwrap_or_else(Utc::now); + let time_range_end = inputs + .iter() + .map(|i| i.time_range_end) + .max() + .unwrap_or_else(Utc::now); + let score = inputs + .iter() + .map(|i| i.score) + .fold(f32::NEG_INFINITY, f32::max) + .max(0.0); + + let ctx = SummaryContext { + tree_id: &tree.id, + tree_kind: TreeKind::Global, + target_level, + token_budget: GLOBAL_TOKEN_BUDGET, + }; + let output = summariser + .summarise(&inputs, &ctx) + .await + .context("summariser failed during global seal")?; + + let now = Utc::now(); + let summary_id = new_summary_id(target_level); + let node = SummaryNode { + id: summary_id.clone(), + tree_id: tree.id.clone(), + tree_kind: TreeKind::Global, + level: target_level, + parent_id: None, + child_ids: buf.item_ids.clone(), + content: output.content, + token_count: output.token_count, + entities: output.entities, + topics: output.topics, + time_range_start, + time_range_end, + score, + sealed_at: now, + deleted: false, + }; + + // Single write transaction: insert the new summary, clear this level's + // buffer, append the new id to the parent buffer, and bump the tree's + // max_level/root_id if we just climbed. Re-read `max_level` inside the + // tx so cascading seals within one call see the bump from earlier + // iterations. + let summary_id_for_closure = summary_id.clone(); + let target_level_for_closure = target_level; + let tree_id = tree.id.clone(); + with_connection(config, move |conn| { + let tx = conn.unchecked_transaction()?; + + let current_max: u32 = tx + .query_row( + "SELECT max_level FROM mem_tree_trees WHERE id = ?1", + rusqlite::params![&tree_id], + |r| r.get::<_, i64>(0), + ) + .map(|n| n.max(0) as u32) + .context("Failed to read current max_level for global tree")?; + + store::insert_summary_tx(&tx, &node)?; + // Index any entities the summariser emitted. No-op under + // InertSummariser (entities stays empty by design — see + // summariser/inert.rs). Becomes active when the Ollama summariser + // lands and emits curated canonical ids. + crate::openhuman::memory::tree::score::store::index_summary_entity_ids_tx( + &tx, + &node.entities, + &node.id, + node.score, + now.timestamp_millis(), + Some(&tree_id), + )?; + // Backlink children → new parent. In the global tree every level is + // already a summary, so the backlink always targets + // `mem_tree_summaries`. + for child_id in &node.child_ids { + tx.execute( + "UPDATE mem_tree_summaries + SET parent_id = ?1 + WHERE id = ?2 AND parent_id IS NULL", + rusqlite::params![&summary_id_for_closure, child_id], + ) + .context("Failed to backlink global summary to parent summary")?; + } + store::clear_buffer_tx(&tx, &tree_id, level)?; + + // Append to parent buffer. + let mut parent = store::get_buffer_conn(&tx, &tree_id, target_level_for_closure)?; + parent.item_ids.push(summary_id_for_closure.clone()); + parent.token_sum = parent.token_sum.saturating_add(node.token_count as i64); + parent.oldest_at = match parent.oldest_at { + Some(existing) => Some(existing.min(time_range_start)), + None => Some(time_range_start), + }; + store::upsert_buffer_tx(&tx, &parent)?; + + // Update tree root / max_level if we just climbed. + if target_level_for_closure > current_max { + store::update_tree_after_seal_tx( + &tx, + &tree_id, + &summary_id_for_closure, + target_level_for_closure, + now, + )?; + } else { + // Same max level — refresh last_sealed_at only. + tx.execute( + "UPDATE mem_tree_trees SET last_sealed_at_ms = ?1 WHERE id = ?2", + rusqlite::params![now.timestamp_millis(), &tree_id], + ) + .context("Failed to refresh last_sealed_at for global tree")?; + } + + tx.commit()?; + Ok(()) + })?; + + log::info!( + "[global_tree::seal] sealed tree_id={} level={}→{} summary_id={} children={}", + tree.id, + level, + target_level, + summary_id, + buf.item_ids.len() + ); + + Ok(summary_id) +} + +/// Hydrate summary rows for the ids in a buffer. Global-tree buffers at +/// every level reference summary nodes (not chunks), so we always pull from +/// `mem_tree_summaries`. +fn hydrate_summary_inputs(config: &Config, summary_ids: &[String]) -> Result> { + let mut out: Vec = Vec::with_capacity(summary_ids.len()); + for id in summary_ids { + let node = match store::get_summary(config, id)? { + Some(n) => n, + None => { + log::warn!( + "[global_tree::seal] hydrate_summary_inputs: missing summary {id} — skipping" + ); + continue; + } + }; + out.push(SummaryInput { + id: node.id.clone(), + content: node.content.clone(), + token_count: node.token_count, + entities: node.entities.clone(), + topics: node.topics.clone(), + time_range_start: node.time_range_start, + time_range_end: node.time_range_end, + score: node.score, + }); + } + Ok(out) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::openhuman::memory::tree::global_tree::registry::get_or_create_global_tree; + use crate::openhuman::memory::tree::source_tree::summariser::inert::InertSummariser; + use chrono::TimeZone; + use tempfile::TempDir; + + fn test_config() -> (TempDir, Config) { + let tmp = TempDir::new().unwrap(); + let mut cfg = Config::default(); + cfg.workspace_dir = tmp.path().to_path_buf(); + (tmp, cfg) + } + + fn mk_daily(id: &str, tree_id: &str, day_ms: i64) -> SummaryNode { + let ts = Utc.timestamp_millis_opt(day_ms).single().unwrap(); + SummaryNode { + id: id.to_string(), + tree_id: tree_id.to_string(), + tree_kind: TreeKind::Global, + level: 0, + parent_id: None, + child_ids: vec![], // not used by seal hydrator + content: format!("daily digest {id}"), + token_count: 200, + entities: vec![], + topics: vec![], + time_range_start: ts, + time_range_end: ts, + score: 0.5, + sealed_at: ts, + deleted: false, + } + } + + fn insert_daily(cfg: &Config, node: &SummaryNode) { + with_connection(cfg, |conn| { + let tx = conn.unchecked_transaction()?; + store::insert_summary_tx(&tx, node)?; + tx.commit()?; + Ok(()) + }) + .unwrap(); + } + + #[tokio::test] + async fn below_threshold_does_not_seal() { + let (_tmp, cfg) = test_config(); + let tree = get_or_create_global_tree(&cfg).unwrap(); + let summariser = InertSummariser::new(); + + // Append 3 daily nodes — well below the 7-day weekly threshold. + for i in 0..3 { + let node = mk_daily( + &format!("summary:L0:day{i}"), + &tree.id, + 1_700_000_000_000 + i, + ); + insert_daily(&cfg, &node); + let sealed = append_daily_and_cascade(&cfg, &tree, &node, &summariser) + .await + .unwrap(); + assert!(sealed.is_empty(), "no cascade expected below threshold"); + } + + let buf = store::get_buffer(&cfg, &tree.id, 0).unwrap(); + assert_eq!(buf.item_ids.len(), 3); + } + + #[tokio::test] + async fn crossing_weekly_threshold_seals_l1() { + let (_tmp, cfg) = test_config(); + let tree = get_or_create_global_tree(&cfg).unwrap(); + let summariser = InertSummariser::new(); + + // Append exactly 7 daily nodes — should trigger one L0→L1 seal. + for i in 0..WEEKLY_SEAL_THRESHOLD { + let node = mk_daily( + &format!("summary:L0:day{i}"), + &tree.id, + 1_700_000_000_000 + i as i64, + ); + insert_daily(&cfg, &node); + let sealed = append_daily_and_cascade(&cfg, &tree, &node, &summariser) + .await + .unwrap(); + if i + 1 < WEEKLY_SEAL_THRESHOLD { + assert!(sealed.is_empty(), "no seal before threshold (i={i})"); + } else { + assert_eq!(sealed.len(), 1, "expected one weekly seal on 7th append"); + } + } + + // L0 buffer cleared; L1 buffer holds the new weekly summary. + let l0 = store::get_buffer(&cfg, &tree.id, 0).unwrap(); + assert!(l0.is_empty()); + let l1 = store::get_buffer(&cfg, &tree.id, 1).unwrap(); + assert_eq!(l1.item_ids.len(), 1); + + // Tree metadata reflects the climb to level 1. + let t = store::get_tree(&cfg, &tree.id).unwrap().unwrap(); + assert_eq!(t.max_level, 1); + assert_eq!(t.root_id.as_deref(), Some(l1.item_ids[0].as_str())); + assert!(t.last_sealed_at.is_some()); + + // Weekly summary row carries children = the 7 daily ids. + let weekly = store::get_summary(&cfg, &l1.item_ids[0]).unwrap().unwrap(); + assert_eq!(weekly.level, 1); + assert_eq!(weekly.tree_kind, TreeKind::Global); + assert_eq!(weekly.child_ids.len(), WEEKLY_SEAL_THRESHOLD); + } + + #[tokio::test] + async fn append_is_idempotent_on_retry() { + let (_tmp, cfg) = test_config(); + let tree = get_or_create_global_tree(&cfg).unwrap(); + let summariser = InertSummariser::new(); + + let node = mk_daily("summary:L0:dayA", &tree.id, 1_700_000_000_000); + insert_daily(&cfg, &node); + append_daily_and_cascade(&cfg, &tree, &node, &summariser) + .await + .unwrap(); + append_daily_and_cascade(&cfg, &tree, &node, &summariser) + .await + .unwrap(); + + let buf = store::get_buffer(&cfg, &tree.id, 0).unwrap(); + assert_eq!( + buf.item_ids.len(), + 1, + "retry must not double-insert the same daily id" + ); + assert_eq!(buf.token_sum, 200); + } +} diff --git a/src/openhuman/memory/tree/mod.rs b/src/openhuman/memory/tree/mod.rs index 742d15034..ba5d8037d 100644 --- a/src/openhuman/memory/tree/mod.rs +++ b/src/openhuman/memory/tree/mod.rs @@ -24,6 +24,7 @@ pub mod canonicalize; pub mod chunker; +pub mod global_tree; pub mod ingest; pub mod rpc; pub mod schemas; diff --git a/src/openhuman/memory/tree/source_tree/store.rs b/src/openhuman/memory/tree/source_tree/store.rs index 82b57e2b7..5c018986f 100644 --- a/src/openhuman/memory/tree/source_tree/store.rs +++ b/src/openhuman/memory/tree/source_tree/store.rs @@ -97,6 +97,26 @@ pub fn get_tree(config: &Config, id: &str) -> Result> { }) } +/// List every tree of a given kind. Used by the global digest to enumerate +/// source trees, and by diagnostics. Rows come back ordered by `created_at_ms` +/// ASC so callers see a stable iteration order. +pub fn list_trees_by_kind(config: &Config, kind: TreeKind) -> Result> { + with_connection(config, |conn| { + let mut stmt = conn.prepare( + "SELECT id, kind, scope, root_id, max_level, status, + created_at_ms, last_sealed_at_ms + FROM mem_tree_trees + WHERE kind = ?1 + ORDER BY created_at_ms ASC", + )?; + let rows = stmt + .query_map(params![kind.as_str()], row_to_tree)? + .collect::>>() + .context("Failed to collect trees by kind")?; + Ok(rows) + }) +} + pub(crate) fn update_tree_after_seal_tx( tx: &Transaction<'_>, tree_id: &str,