From 8e55ddd71e8d80b881a4d8921e1ebeb260ec567b Mon Sep 17 00:00:00 2001 From: sanil-23 Date: Wed, 22 Apr 2026 20:26:31 +0530 Subject: [PATCH] feat(memory): source trees + bucket-seal foundation (#709) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Phase 3a of the memory architecture (#711 umbrella). Lifts admitted leaves into a per-source hierarchy so Phase 4 retrieval has a tree to walk. ## What's in this PR * **Schema** — three new tables in `chunks.db` alongside Phase 1/2: `mem_tree_trees`, `mem_tree_summaries`, `mem_tree_buffers`. Plus a `parent_summary_id` backlink column on `mem_tree_chunks`. All migrations are additive and idempotent (`ALTER TABLE ADD COLUMN`, same pattern as Phase 2). * **`source_tree/` module** — isolated subdir; does not touch the legacy `openhuman::memory` layer. - `types.rs` — `Tree`, `SummaryNode`, `Buffer`, `TreeKind` (Source/Topic/Global), `TreeStatus`. Constants: `TOKEN_BUDGET = 10_000`, `DEFAULT_FLUSH_AGE_SECS = 7d`. - `store.rs` — CRUD for all three tables via the shared `with_connection` entry point. Writes are transactional; summary insert is idempotent on primary key (INSERT OR IGNORE). - `registry.rs` — `get_or_create_source_tree(scope)` — idempotent lookup keyed on `(kind, scope)`. - `bucket_seal.rs` — `append_leaf` + cascade: push a leaf into L0, seal when `token_sum >= TOKEN_BUDGET`, recurse upward. One seal = one transaction; children get a parent backlink (leaves via `parent_summary_id`, summaries via `parent_id`). Tree's `max_level` and `root_id` bump on root split. - `flush.rs` — `flush_stale_buffers(max_age)` force-seals any buffer whose `oldest_at` is older than `max_age`. Primitive only; wiring into a scheduler is out of scope. - `summariser/` — `Summariser` trait + `InertSummariser` fallback (concatenate children, union entities preserving first-seen order, hard-truncate to budget). Trait is async; Ollama implementation slots in later without breaking callers. * **Ingest wiring** — `tree/ingest.rs::persist` now calls `append_leaf` once per kept chunk after chunks + scores land. Failures are logged at warn level and do not fail the ingest — leaves are already persisted and the next flush can still rebuild the tree. ## Concurrency / LLM The async summariser call happens OUTSIDE any DB transaction, so a slow LLM never holds SQLite locks. Blocking DB calls inside `append_leaf` are acceptable for Phase 3a because the Inert summariser does no real I/O; when a networked summariser lands, wrap the DB calls in `tokio::task::spawn_blocking`. ## Tests (23 new, all green) * `types` — enum round-trips, buffer staleness predicates * `store` — tree insert + unique scope, summary insert + idempotence, buffer upsert/clear, stale-buffer ordering * `registry` — `get_or_create` idempotence, distinct scopes yield distinct trees, id prefix format * `summariser/inert` — provenance-prefixed concat, entity union with order preservation, budget truncation, empty-content skip * `bucket_seal` — append-below-budget is buffered only (no seal); 12k-token cross-over produces an L1 summary with correct `child_ids`, L0 cleared, L1 carries the new summary id, tree `max_level`/`root_id` updated, leaf → parent backlink populated * `flush` — stale buffer force-seals under budget; recent buffer is a no-op Broader `memory::tree` suite: 160 tests pass (23 new + 137 existing Phase 1/2), no regressions. ## Stacked on #775 This PR applies on top of `feat/708-memory-llm-ner` (PR #775, Phase 2 LLM-NER follow-up). Merge #775 first; this branch will rebase cleanly onto main. ## Out of scope (tracked separately) * **3b — Global activity digest tree** — time-aligned cross-source recap * **3c — Topic trees** — hotness-driven per-entity materialisation * **Phase 4 (#710)** — retrieval / query / gate Closes part of #709 (source-tree foundation). Co-Authored-By: Claude Opus 4.7 (1M context) --- src/openhuman/memory/tree/ingest.rs | 66 ++ src/openhuman/memory/tree/mod.rs | 1 + .../memory/tree/source_tree/bucket_seal.rs | 545 ++++++++++++++++ .../memory/tree/source_tree/flush.rs | 192 ++++++ src/openhuman/memory/tree/source_tree/mod.rs | 28 + .../memory/tree/source_tree/registry.rs | 99 +++ .../memory/tree/source_tree/store.rs | 610 ++++++++++++++++++ .../tree/source_tree/summariser/inert.rs | 190 ++++++ .../memory/tree/source_tree/summariser/mod.rs | 62 ++ .../memory/tree/source_tree/types.rs | 215 ++++++ src/openhuman/memory/tree/store.rs | 69 ++ 11 files changed, 2077 insertions(+) create mode 100644 src/openhuman/memory/tree/source_tree/bucket_seal.rs create mode 100644 src/openhuman/memory/tree/source_tree/flush.rs create mode 100644 src/openhuman/memory/tree/source_tree/mod.rs create mode 100644 src/openhuman/memory/tree/source_tree/registry.rs create mode 100644 src/openhuman/memory/tree/source_tree/store.rs create mode 100644 src/openhuman/memory/tree/source_tree/summariser/inert.rs create mode 100644 src/openhuman/memory/tree/source_tree/summariser/mod.rs create mode 100644 src/openhuman/memory/tree/source_tree/types.rs diff --git a/src/openhuman/memory/tree/ingest.rs b/src/openhuman/memory/tree/ingest.rs index 4b3cbeeb3..448a41566 100644 --- a/src/openhuman/memory/tree/ingest.rs +++ b/src/openhuman/memory/tree/ingest.rs @@ -20,6 +20,9 @@ use crate::openhuman::memory::tree::canonicalize::{ }; use crate::openhuman::memory::tree::chunker::{chunk_markdown, ChunkerInput, ChunkerOptions}; use crate::openhuman::memory::tree::score::{self, ScoreResult, ScoringConfig}; +use crate::openhuman::memory::tree::source_tree::{ + append_leaf, get_or_create_source_tree, InertSummariser, LeafRef, +}; use crate::openhuman::memory::tree::store; use crate::openhuman::memory::tree::types::Chunk; @@ -182,6 +185,20 @@ async fn persist( .await .map_err(|e| anyhow::anyhow!("persist join error: {e}"))??; + // 5. Source-tree append (Phase 3a #709). Each kept leaf pushes into + // the tree's L0 buffer and cascades upward when token_sum crosses + // the budget. Entities/topics from the scorer are threaded in so + // sealed summaries inherit the child signal set. Failures here + // log at warn level but don't fail the ingest — leaves are already + // persisted, and a later flush/retry can still rebuild the tree. + if let Err(e) = append_leaves_to_tree(config, source_id, &kept_chunks, &all_results).await { + log::warn!( + "[memory_tree::ingest] source_tree append failed source_id={} err={:#}", + source_id, + e + ); + } + Ok(IngestResult { source_id: source_id.to_string(), chunks_written: written, @@ -190,6 +207,55 @@ async fn persist( }) } +/// Push every kept chunk into its source tree. Scoped to Phase 3a — all +/// chunks from one ingest batch share the same `source_id`, so they share +/// one tree lookup. The inert summariser is the default; future wiring +/// can swap in an LLM summariser here. +async fn append_leaves_to_tree( + config: &Config, + source_id: &str, + kept_chunks: &[Chunk], + all_results: &[(ScoreResult, i64)], +) -> Result<()> { + if kept_chunks.is_empty() { + return Ok(()); + } + let tree = get_or_create_source_tree(config, source_id)?; + let summariser = InertSummariser::new(); + + // Build a chunk_id → (score, entities, topics) map for quick lookup. + use std::collections::HashMap; + let mut score_by_id: HashMap = HashMap::new(); + for (r, _) in all_results { + score_by_id.insert(r.chunk_id.clone(), r); + } + + for chunk in kept_chunks { + let (score_value, entities, topics) = match score_by_id.get(&chunk.id) { + Some(r) => ( + r.total, + r.canonical_entities + .iter() + .map(|e| e.canonical_id.clone()) + .collect(), + chunk.metadata.tags.clone(), + ), + None => (0.0, Vec::new(), chunk.metadata.tags.clone()), + }; + let leaf = LeafRef { + chunk_id: chunk.id.clone(), + token_count: chunk.token_count, + timestamp: chunk.metadata.timestamp, + content: chunk.content.clone(), + entities, + topics, + score: score_value, + }; + append_leaf(config, &tree, &leaf, &summariser).await?; + } + Ok(()) +} + #[cfg(test)] mod tests { use super::*; diff --git a/src/openhuman/memory/tree/mod.rs b/src/openhuman/memory/tree/mod.rs index d8b157141..742d15034 100644 --- a/src/openhuman/memory/tree/mod.rs +++ b/src/openhuman/memory/tree/mod.rs @@ -28,6 +28,7 @@ pub mod ingest; pub mod rpc; pub mod schemas; pub mod score; +pub mod source_tree; pub mod store; pub mod types; diff --git a/src/openhuman/memory/tree/source_tree/bucket_seal.rs b/src/openhuman/memory/tree/source_tree/bucket_seal.rs new file mode 100644 index 000000000..0ff9d30e0 --- /dev/null +++ b/src/openhuman/memory/tree/source_tree/bucket_seal.rs @@ -0,0 +1,545 @@ +//! Append + cascade-seal for summary trees (#709). +//! +//! `append_leaf` pushes a persisted chunk into the L0 buffer of a tree. If +//! the buffer's running `token_sum` crosses `TOKEN_BUDGET`, the buffer +//! seals into a level-1 summary, its items move into the summary's +//! `child_ids`, the buffer clears, and the new summary id is queued at +//! level 2. The cascade continues upward until a buffer stays under the +//! token budget. +//! +//! Concurrency: Phase 3a assumes a single-process SQLite workspace. All +//! writes in one seal step run in a single transaction; the async +//! summariser call happens outside any open transaction so a slow LLM +//! doesn't hold DB locks. Callers should serialise `append_leaf` per +//! tree — ingest achieves this by processing one batch's chunks +//! sequentially inside its `persist` task. Blocking SQLite calls inside +//! this async function are acceptable for Phase 3a because the Inert +//! summariser does no real I/O; when a networked summariser lands, wrap +//! DB calls in `tokio::task::spawn_blocking` to keep the runtime healthy. + +use anyhow::{Context, Result}; +use chrono::{DateTime, Utc}; +use rusqlite::Transaction; + +use crate::openhuman::config::Config; +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::{ + SummaryContext, SummaryInput, Summariser, +}; +use crate::openhuman::memory::tree::source_tree::types::{ + Buffer, SummaryNode, Tree, TreeKind, TOKEN_BUDGET, +}; +use crate::openhuman::memory::tree::store::with_connection; + +/// Hard cap on cascade depth — prevents runaway loops if token accounting +/// ever slips. 32 levels at even a 2x fan-in is more than enough for any +/// realistic source. +const MAX_CASCADE_DEPTH: u32 = 32; + +/// A single leaf being appended to an L0 buffer. +#[derive(Clone, Debug)] +pub struct LeafRef { + pub chunk_id: String, + pub token_count: u32, + pub timestamp: DateTime, + pub content: String, + pub entities: Vec, + pub topics: Vec, + pub score: f32, +} + +/// Append a leaf to the source tree for `tree`, sealing buffers as they +/// fill. Returns the ids of any summaries that sealed during this call. +pub async fn append_leaf( + config: &Config, + tree: &Tree, + leaf: &LeafRef, + summariser: &dyn Summariser, +) -> Result> { + log::debug!( + "[source_tree::bucket_seal] append_leaf tree_id={} leaf_id={} tokens={}", + tree.id, + leaf.chunk_id, + leaf.token_count + ); + + // 1. Push leaf into L0 buffer (transactional). + append_to_buffer( + config, + &tree.id, + 0, + &leaf.chunk_id, + leaf.token_count as i64, + leaf.timestamp, + )?; + + // 2. Cascade seals upward until a level stays under budget. + cascade_seals(config, tree, summariser).await +} + +/// Transactionally append a single item to `(tree_id, level)`'s buffer. +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)?; + 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> { + cascade_all_from(config, tree, 0, summariser, None).await +} + +/// Seal buffers starting at `start_level` and cascade upward. When +/// `force_now` is `Some`, the buffer at `start_level` is sealed regardless +/// of token budget (used by time-based flush). Upper levels are sealed +/// only when they cross the budget. +pub async fn cascade_all_from( + config: &Config, + tree: &Tree, + start_level: u32, + summariser: &dyn Summariser, + force_now: Option>, +) -> Result> { + let mut sealed_ids: Vec = Vec::new(); + let mut level: u32 = start_level; + let mut first_iteration = true; + + for _ in 0..MAX_CASCADE_DEPTH { + let buf = store::get_buffer(config, &tree.id, level)?; + let forced = first_iteration && force_now.is_some(); + first_iteration = false; + + if !forced && !should_seal(&buf) { + log::debug!( + "[source_tree::bucket_seal] cascade done tree_id={} stop_level={} token_sum={}", + tree.id, + level, + buf.token_sum + ); + break; + } + if buf.is_empty() { + log::debug!( + "[source_tree::bucket_seal] cascade hit empty buffer tree_id={} level={} — stopping", + tree.id, + level + ); + break; + } + + let summary_id = seal_one_level(config, tree, &buf, summariser).await?; + sealed_ids.push(summary_id); + level += 1; + } + + Ok(sealed_ids) +} + +fn should_seal(buf: &Buffer) -> bool { + !buf.is_empty() && buf.token_sum >= TOKEN_BUDGET as i64 +} + +/// Seal `buf` at `level` into one summary at `level + 1`. Returns the new +/// summary id. +async fn seal_one_level( + config: &Config, + tree: &Tree, + buf: &Buffer, + summariser: &dyn Summariser, +) -> Result { + let level = buf.level; + let target_level = level + 1; + + // Hydrate inputs (synchronous DB reads). + let inputs = hydrate_inputs(config, level, &buf.item_ids)?; + if inputs.is_empty() { + anyhow::bail!( + "[source_tree::bucket_seal] refused to seal empty buffer tree_id={} level={}", + tree.id, + level + ); + } + + // Compute envelope across children (time range, max score). + 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); + + // Run summariser — async, OUTSIDE any DB transaction. + let ctx = SummaryContext { + tree_id: &tree.id, + tree_kind: TreeKind::Source, + target_level, + token_budget: TOKEN_BUDGET, + }; + let output = summariser + .summarise(&inputs, &ctx) + .await + .context("summariser failed during seal")?; + + // Build the new summary node. + 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::Source, + 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 summary, clear this buffer, append + // summary id to parent buffer, bump tree max_level/root if needed. + // Re-read `max_level` from inside the tx so cascading seals within + // one call see the updated value from earlier levels. + 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 tree")?; + + store::insert_summary_tx(&tx, &node)?; + // Backlink children → new parent so leaf/parent traversal is a + // single-row lookup in Phase 4. Skipped for levels already bound + // to a parent (shouldn't happen — a child seals at most once). + for child_id in &node.child_ids { + if level == 0 { + tx.execute( + "UPDATE mem_tree_chunks + SET parent_summary_id = ?1 + WHERE id = ?2 AND parent_summary_id IS NULL", + rusqlite::params![&summary_id_for_closure, child_id], + ) + .context("Failed to backlink chunk to parent summary")?; + } else { + 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 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 — still refresh last_sealed_at via same helper + // but keep existing root intact. Root tracking at the same + // level is resolved in Phase 4 retrieval. + refresh_last_sealed_tx(&tx, &tree_id, now)?; + } + + tx.commit()?; + Ok(()) + })?; + + log::info!( + "[source_tree::bucket_seal] sealed tree_id={} level={}→{} summary_id={} children={}", + tree.id, + level, + target_level, + summary_id, + buf.item_ids.len() + ); + + Ok(summary_id) +} + +fn refresh_last_sealed_tx( + tx: &Transaction<'_>, + tree_id: &str, + sealed_at: DateTime, +) -> Result<()> { + tx.execute( + "UPDATE mem_tree_trees SET last_sealed_at_ms = ?1 WHERE id = ?2", + rusqlite::params![sealed_at.timestamp_millis(), tree_id], + ) + .with_context(|| format!("Failed to refresh last_sealed_at for tree {tree_id}"))?; + Ok(()) +} + +/// 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`. +fn hydrate_inputs( + config: &Config, + level: u32, + item_ids: &[String], +) -> Result> { + if level == 0 { + hydrate_leaf_inputs(config, item_ids) + } else { + hydrate_summary_inputs(config, item_ids) + } +} + +fn hydrate_leaf_inputs(config: &Config, chunk_ids: &[String]) -> Result> { + use crate::openhuman::memory::tree::score::store::get_score; + use crate::openhuman::memory::tree::store::get_chunk; + + let mut out: Vec = Vec::with_capacity(chunk_ids.len()); + for id in chunk_ids { + let chunk = match get_chunk(config, id)? { + Some(c) => c, + None => { + log::warn!( + "[source_tree::bucket_seal] hydrate_leaf_inputs: missing chunk {id} — skipping" + ); + continue; + } + }; + let score = get_score(config, id)?; + let (score_value, entities, topics) = match &score { + Some(row) => (row.total, Vec::new(), chunk.metadata.tags.clone()), + None => (0.0, Vec::new(), chunk.metadata.tags.clone()), + }; + out.push(SummaryInput { + id: chunk.id.clone(), + content: chunk.content.clone(), + token_count: chunk.token_count, + entities, + topics, + time_range_start: chunk.metadata.time_range.0, + time_range_end: chunk.metadata.time_range.1, + score: score_value, + }); + } + Ok(out) +} + +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!( + "[source_tree::bucket_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::source_tree::registry::get_or_create_source_tree; + use crate::openhuman::memory::tree::source_tree::summariser::inert::InertSummariser; + 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_leaf(id: &str, tokens: u32, ts_ms: i64) -> LeafRef { + use chrono::TimeZone; + LeafRef { + chunk_id: id.to_string(), + token_count: tokens, + timestamp: Utc.timestamp_millis_opt(ts_ms).single().unwrap(), + content: format!("content for {id}"), + entities: vec![], + topics: vec![], + score: 0.5, + } + } + + #[tokio::test] + async fn append_below_budget_does_not_seal() { + let (_tmp, cfg) = test_config(); + let tree = get_or_create_source_tree(&cfg, "slack:#eng").unwrap(); + let summariser = InertSummariser::new(); + // Chunks don't exist in DB — we're only exercising the buffer + // accounting, which doesn't require leaf rows until a seal fires. + let leaf = mk_leaf("leaf-1", 100, 1_700_000_000_000); + let sealed = append_leaf(&cfg, &tree, &leaf, &summariser).await.unwrap(); + assert!(sealed.is_empty()); + + let buf = store::get_buffer(&cfg, &tree.id, 0).unwrap(); + assert_eq!(buf.item_ids, vec!["leaf-1".to_string()]); + assert_eq!(buf.token_sum, 100); + assert_eq!(store::count_summaries(&cfg, &tree.id).unwrap(), 0); + } + + #[tokio::test] + async fn crossing_budget_triggers_seal() { + use crate::openhuman::memory::tree::store::upsert_chunks; + use crate::openhuman::memory::tree::types::{ + chunk_id, Chunk, Metadata, SourceKind, SourceRef, + }; + use chrono::TimeZone; + + let (_tmp, cfg) = test_config(); + let tree = get_or_create_source_tree(&cfg, "slack:#eng").unwrap(); + let summariser = InertSummariser::new(); + + // Persist two chunks that the hydrator can load during seal. + let ts = Utc.timestamp_millis_opt(1_700_000_000_000).unwrap(); + let mk_chunk = |seq: u32, tokens: u32| Chunk { + id: chunk_id(SourceKind::Chat, "slack:#eng", seq), + content: format!("substantive chunk content {seq}"), + metadata: Metadata { + source_kind: SourceKind::Chat, + source_id: "slack:#eng".into(), + owner: "alice".into(), + timestamp: ts, + time_range: (ts, ts), + tags: vec![], + source_ref: Some(SourceRef::new("slack://x")), + }, + token_count: tokens, + seq_in_source: seq, + created_at: ts, + }; + let c1 = mk_chunk(0, 6_000); + let c2 = mk_chunk(1, 6_000); + upsert_chunks(&cfg, &[c1.clone(), c2.clone()]).unwrap(); + + // Two leaves whose combined token_sum (12k) exceeds the 10k budget. + 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, + }; + + let first = append_leaf(&cfg, &tree, &leaf1, &summariser).await.unwrap(); + assert!(first.is_empty(), "first append below budget — no seal"); + + let second = append_leaf(&cfg, &tree, &leaf2, &summariser).await.unwrap(); + assert_eq!(second.len(), 1, "second append crosses budget — one seal"); + + let summary_id = &second[0]; + let summary = store::get_summary(&cfg, summary_id).unwrap().unwrap(); + assert_eq!(summary.level, 1); + assert_eq!(summary.child_ids, vec![c1.id.clone(), c2.id.clone()]); + assert!(summary.token_count > 0); + + // L0 buffer cleared, L1 buffer carries the new summary id. + 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, vec![summary_id.clone()]); + + // Tree metadata updated. + let t = store::get_tree(&cfg, &tree.id).unwrap().unwrap(); + assert_eq!(t.max_level, 1); + assert_eq!(t.root_id.as_deref(), Some(summary_id.as_str())); + assert!(t.last_sealed_at.is_some()); + + // Leaf → parent backlink populated for both children. + use crate::openhuman::memory::tree::store::with_connection; + let parent: Option = with_connection(&cfg, |conn| { + let p: Option = conn + .query_row( + "SELECT parent_summary_id FROM mem_tree_chunks WHERE id = ?1", + rusqlite::params![c1.id], + |r| r.get(0), + ) + .unwrap(); + Ok(p) + }) + .unwrap(); + assert_eq!(parent.as_deref(), Some(summary_id.as_str())); + } +} diff --git a/src/openhuman/memory/tree/source_tree/flush.rs b/src/openhuman/memory/tree/source_tree/flush.rs new file mode 100644 index 000000000..d545fb58e --- /dev/null +++ b/src/openhuman/memory/tree/source_tree/flush.rs @@ -0,0 +1,192 @@ +//! Time-based buffer flush for source trees (#709). +//! +//! The bucket-seal path only fires when a buffer crosses `TOKEN_BUDGET`. +//! Low-volume sources (e.g. an email account with two threads a week) can +//! otherwise leave leaves parked in the L0 buffer indefinitely, which +//! hurts recall. `flush_stale_buffers` force-seals any buffer whose +//! `oldest_at` is older than `max_age`, regardless of token count. +//! +//! This is meant to run on a cadence (e.g. daily cron). Phase 3a ships +//! the primitive; wiring into a scheduler is not required for merge. + +use anyhow::Result; +use chrono::{DateTime, Duration, Utc}; + +use crate::openhuman::config::Config; +use crate::openhuman::memory::tree::source_tree::bucket_seal::cascade_all_from; +use crate::openhuman::memory::tree::source_tree::store; +use crate::openhuman::memory::tree::source_tree::summariser::Summariser; +use crate::openhuman::memory::tree::source_tree::types::DEFAULT_FLUSH_AGE_SECS; + +/// Seal every buffer whose oldest item is older than `max_age`. Returns +/// the number of individual seal calls (not trees) that fired. When the +/// same tree has multiple stale levels they're each sealed in order. +pub async fn flush_stale_buffers( + config: &Config, + max_age: Duration, + summariser: &dyn Summariser, +) -> Result { + let now = Utc::now(); + let cutoff = now - max_age; + let stale = store::list_stale_buffers(config, cutoff)?; + log::info!( + "[source_tree::flush] found {} stale buffers (max_age={:?})", + stale.len(), + max_age + ); + + let mut seals: usize = 0; + for buf in stale { + let tree = match store::get_tree(config, &buf.tree_id)? { + Some(t) => t, + None => { + log::warn!( + "[source_tree::flush] orphan buffer tree_id={} level={}", + buf.tree_id, + buf.level + ); + continue; + } + }; + let sealed = cascade_all_from(config, &tree, buf.level, summariser, Some(now)).await?; + seals += sealed.len(); + } + Ok(seals) +} + +/// Convenience wrapper that uses [`DEFAULT_FLUSH_AGE_SECS`]. +pub async fn flush_stale_buffers_default( + config: &Config, + summariser: &dyn Summariser, +) -> Result { + flush_stale_buffers( + config, + Duration::seconds(DEFAULT_FLUSH_AGE_SECS), + summariser, + ) + .await +} + +/// Helper exposed for callers that want a single explicit force-seal (e.g. +/// "user disconnected this account, flush its buffer now"). +pub async fn force_flush_tree( + config: &Config, + tree_id: &str, + summariser: &dyn Summariser, + now: Option>, +) -> Result> { + let tree = store::get_tree(config, tree_id)? + .ok_or_else(|| anyhow::anyhow!("no tree with id {tree_id}"))?; + cascade_all_from(config, &tree, 0, summariser, now).await +} + +#[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::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) + } + + #[tokio::test] + async fn flush_seals_old_buffer_even_under_budget() { + let (_tmp, cfg) = test_config(); + let tree = get_or_create_source_tree(&cfg, "slack:#eng").unwrap(); + let summariser = InertSummariser::new(); + + // Persist one chunk with an old timestamp (10 days ago). + let old_ts = Utc::now() - Duration::days(10); + let c = Chunk { + id: chunk_id(SourceKind::Chat, "slack:#eng", 0), + content: "old content that should get sealed".into(), + metadata: Metadata { + source_kind: SourceKind::Chat, + source_id: "slack:#eng".into(), + owner: "alice".into(), + timestamp: old_ts, + time_range: (old_ts, old_ts), + tags: vec![], + source_ref: Some(SourceRef::new("slack://x")), + }, + token_count: 100, + seq_in_source: 0, + created_at: old_ts, + }; + upsert_chunks(&cfg, &[c.clone()]).unwrap(); + + let leaf = LeafRef { + chunk_id: c.id.clone(), + token_count: 100, // way under the 10k budget + timestamp: old_ts, + content: c.content.clone(), + entities: vec![], + topics: vec![], + score: 0.5, + }; + append_leaf(&cfg, &tree, &leaf, &summariser).await.unwrap(); + assert_eq!(store::count_summaries(&cfg, &tree.id).unwrap(), 0); + + let seals = flush_stale_buffers(&cfg, Duration::days(7), &summariser) + .await + .unwrap(); + assert_eq!(seals, 1); + assert_eq!(store::count_summaries(&cfg, &tree.id).unwrap(), 1); + + let l0 = store::get_buffer(&cfg, &tree.id, 0).unwrap(); + assert!(l0.is_empty()); + } + + #[tokio::test] + async fn flush_noop_when_buffer_is_recent() { + let (_tmp, cfg) = test_config(); + let tree = get_or_create_source_tree(&cfg, "slack:#eng").unwrap(); + let summariser = InertSummariser::new(); + + // Persist a leaf stamped now so it's NOT stale. + let now = Utc::now(); + let c = Chunk { + id: chunk_id(SourceKind::Chat, "slack:#eng", 0), + content: "fresh".into(), + metadata: Metadata { + source_kind: SourceKind::Chat, + source_id: "slack:#eng".into(), + owner: "alice".into(), + timestamp: now, + time_range: (now, now), + tags: vec![], + source_ref: Some(SourceRef::new("slack://x")), + }, + token_count: 50, + seq_in_source: 0, + created_at: now, + }; + upsert_chunks(&cfg, &[c.clone()]).unwrap(); + let leaf = LeafRef { + chunk_id: c.id.clone(), + token_count: 50, + timestamp: now, + content: c.content.clone(), + entities: vec![], + topics: vec![], + score: 0.5, + }; + append_leaf(&cfg, &tree, &leaf, &summariser).await.unwrap(); + + let seals = flush_stale_buffers(&cfg, Duration::days(7), &summariser) + .await + .unwrap(); + assert_eq!(seals, 0); + assert_eq!(store::count_summaries(&cfg, &tree.id).unwrap(), 0); + } +} diff --git a/src/openhuman/memory/tree/source_tree/mod.rs b/src/openhuman/memory/tree/source_tree/mod.rs new file mode 100644 index 000000000..8f8d6ed38 --- /dev/null +++ b/src/openhuman/memory/tree/source_tree/mod.rs @@ -0,0 +1,28 @@ +//! Phase 3a — summary trees + bucket-seal mechanics (#709). +//! +//! A thin orchestration layer on top of Phase 1 chunks and Phase 2 scores +//! that lifts individual leaves into a hierarchy of sealed summary nodes, +//! one tree per ingest source. See `docs/MEMORY_ARCHITECTURE_LLD.md` for +//! the full design. The module is isolated from the legacy +//! `openhuman::memory` layer and only depends on sibling `tree::*` modules. +//! +//! Public surface at Phase 3a: +//! - [`registry::get_or_create_source_tree`] — idempotent tree lookup +//! - [`bucket_seal::append_leaf`] — push a chunk into its tree, cascade-seal on budget +//! - [`flush::flush_stale_buffers`] — time-based seal of buffers that never cross budget +//! - [`summariser::inert::InertSummariser`] — deterministic fallback summariser +//! +//! Phases 3b / 3c will add `global` and `topic` trees; both reuse the +//! storage and cascade primitives defined here. + +pub mod bucket_seal; +pub mod flush; +pub mod registry; +pub mod store; +pub mod summariser; +pub mod types; + +pub use bucket_seal::{append_leaf, LeafRef}; +pub use registry::get_or_create_source_tree; +pub use summariser::{inert::InertSummariser, Summariser}; +pub use types::{Buffer, SummaryNode, Tree, TreeKind, TreeStatus, TOKEN_BUDGET}; diff --git a/src/openhuman/memory/tree/source_tree/registry.rs b/src/openhuman/memory/tree/source_tree/registry.rs new file mode 100644 index 000000000..8cef91a12 --- /dev/null +++ b/src/openhuman/memory/tree/source_tree/registry.rs @@ -0,0 +1,99 @@ +//! Tree registry — get-or-create for source trees (#709). +//! +//! The registry is the entry point for the ingest path to look up the +//! tree for a given (kind, scope). Phase 3a only touches source trees; +//! topic / global trees will reuse the same `(kind, scope)` convention +//! in Phases 3b / 3c. + +use anyhow::Result; +use chrono::Utc; +use uuid::Uuid; + +use crate::openhuman::config::Config; +use crate::openhuman::memory::tree::source_tree::store; +use crate::openhuman::memory::tree::source_tree::types::{Tree, TreeKind, TreeStatus}; + +/// Look up the source tree for `scope`, or create a new one. +/// +/// Scope format convention (Phase 3a): use the ingested chunk's +/// `metadata.source_id` verbatim, so re-ingesting the same Slack channel +/// or Gmail account keeps appending to the same tree. +pub fn get_or_create_source_tree(config: &Config, scope: &str) -> Result { + if let Some(existing) = store::get_tree_by_scope(config, TreeKind::Source, scope)? { + log::debug!( + "[source_tree::registry] found tree id={} scope={}", + existing.id, + scope + ); + return Ok(existing); + } + + let tree = Tree { + id: new_tree_id(TreeKind::Source), + kind: TreeKind::Source, + scope: scope.to_string(), + root_id: None, + max_level: 0, + status: TreeStatus::Active, + created_at: Utc::now(), + last_sealed_at: None, + }; + store::insert_tree(config, &tree)?; + log::info!( + "[source_tree::registry] created source tree id={} scope={}", + tree.id, + scope + ); + Ok(tree) +} + +fn new_tree_id(kind: TreeKind) -> String { + format!("{}:{}", kind.as_str(), Uuid::new_v4()) +} + +/// Public id generator for summary nodes — exported so `bucket_seal` can +/// share the same format (kept separate for readability; both use UUID v4 +/// suffixes to keep ids short but unambiguous). +pub fn new_summary_id(level: u32) -> String { + format!("summary:L{}:{}", level, 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_on_scope() { + let (_tmp, cfg) = test_config(); + let first = get_or_create_source_tree(&cfg, "slack:#eng").unwrap(); + let second = get_or_create_source_tree(&cfg, "slack:#eng").unwrap(); + assert_eq!(first.id, second.id); + assert_eq!(first.kind, TreeKind::Source); + assert_eq!(first.status, TreeStatus::Active); + } + + #[test] + fn different_scopes_yield_different_trees() { + let (_tmp, cfg) = test_config(); + let a = get_or_create_source_tree(&cfg, "slack:#eng").unwrap(); + let b = get_or_create_source_tree(&cfg, "gmail:user@example.com").unwrap(); + assert_ne!(a.id, b.id); + assert_ne!(a.scope, b.scope); + } + + #[test] + fn tree_id_has_expected_prefix() { + let id = new_tree_id(TreeKind::Source); + assert!(id.starts_with("source:")); + let sum_id = new_summary_id(3); + assert!(sum_id.starts_with("summary:L3:")); + } +} diff --git a/src/openhuman/memory/tree/source_tree/store.rs b/src/openhuman/memory/tree/source_tree/store.rs new file mode 100644 index 000000000..abeab2853 --- /dev/null +++ b/src/openhuman/memory/tree/source_tree/store.rs @@ -0,0 +1,610 @@ +//! SQLite-backed persistence for Phase 3a summary trees (#709). +//! +//! Three tables (schema lives in the sibling `tree::store::SCHEMA`): +//! - `mem_tree_trees` — one row per tree (kind, scope, root, max_level) +//! - `mem_tree_summaries` — one row per sealed summary node (immutable) +//! - `mem_tree_buffers` — one row per unsealed frontier `(tree_id, level)` +//! +//! All timestamps are stored as milliseconds since the Unix epoch so we +//! share the epoch convention with `mem_tree_chunks`. Writes are serialised +//! through the sibling `tree::store::with_connection` so we inherit its +//! busy-timeout, WAL, and schema-init behaviour. + +use anyhow::{Context, Result}; +use chrono::{DateTime, TimeZone, Utc}; +use rusqlite::{params, Connection, OptionalExtension, Transaction}; + +use crate::openhuman::config::Config; +use crate::openhuman::memory::tree::source_tree::types::{ + Buffer, SummaryNode, Tree, TreeKind, TreeStatus, +}; +use crate::openhuman::memory::tree::store::with_connection; + +fn ms_to_utc(ms: i64) -> rusqlite::Result> { + Utc.timestamp_millis_opt(ms).single().ok_or_else(|| { + rusqlite::Error::FromSqlConversionFailure( + 0, + rusqlite::types::Type::Integer, + format!("invalid timestamp ms {ms}").into(), + ) + }) +} + +// ── Tree rows ─────────────────────────────────────────────────────────── + +/// Insert a new tree row. Fails if `(kind, scope)` already exists; callers +/// that want "get or create" semantics should go through the `registry`. +pub fn insert_tree(config: &Config, tree: &Tree) -> Result<()> { + with_connection(config, |conn| insert_tree_conn(conn, tree)) +} + +pub(crate) fn insert_tree_conn(conn: &Connection, tree: &Tree) -> Result<()> { + conn.execute( + "INSERT INTO mem_tree_trees ( + id, kind, scope, root_id, max_level, status, + created_at_ms, last_sealed_at_ms + ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)", + params![ + tree.id, + tree.kind.as_str(), + tree.scope, + tree.root_id, + tree.max_level, + tree.status.as_str(), + tree.created_at.timestamp_millis(), + tree.last_sealed_at.map(|t| t.timestamp_millis()), + ], + ) + .with_context(|| format!("Failed to insert tree id={}", tree.id))?; + Ok(()) +} + +/// Fetch a tree by `(kind, scope)`. Returns `None` if no such tree exists. +pub fn get_tree_by_scope( + config: &Config, + kind: TreeKind, + scope: &str, +) -> Result> { + with_connection(config, |conn| get_tree_by_scope_conn(conn, kind, scope)) +} + +pub(crate) fn get_tree_by_scope_conn( + conn: &Connection, + kind: TreeKind, + scope: &str, +) -> Result> { + 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 AND scope = ?2", + )?; + let row = stmt + .query_row(params![kind.as_str(), scope], row_to_tree) + .optional() + .context("Failed to query tree by scope")?; + Ok(row) +} + +/// Fetch a tree by primary key id. +pub fn get_tree(config: &Config, id: &str) -> 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 id = ?1", + )?; + let row = stmt + .query_row(params![id], row_to_tree) + .optional() + .context("Failed to query tree by id")?; + Ok(row) + }) +} + +pub(crate) fn update_tree_after_seal_tx( + tx: &Transaction<'_>, + tree_id: &str, + root_id: &str, + max_level: u32, + sealed_at: DateTime, +) -> Result<()> { + tx.execute( + "UPDATE mem_tree_trees + SET root_id = ?1, + max_level = ?2, + last_sealed_at_ms = ?3 + WHERE id = ?4", + params![ + root_id, + max_level, + sealed_at.timestamp_millis(), + tree_id, + ], + ) + .with_context(|| format!("Failed to update tree {tree_id} after seal"))?; + Ok(()) +} + +fn row_to_tree(row: &rusqlite::Row<'_>) -> rusqlite::Result { + let id: String = row.get(0)?; + let kind_s: String = row.get(1)?; + let scope: String = row.get(2)?; + let root_id: Option = row.get(3)?; + let max_level: i64 = row.get(4)?; + let status_s: String = row.get(5)?; + let created_ms: i64 = row.get(6)?; + let last_sealed_ms: Option = row.get(7)?; + + let kind = TreeKind::parse(&kind_s).map_err(|e| { + rusqlite::Error::FromSqlConversionFailure(1, rusqlite::types::Type::Text, e.into()) + })?; + let status = TreeStatus::parse(&status_s).map_err(|e| { + rusqlite::Error::FromSqlConversionFailure(5, rusqlite::types::Type::Text, e.into()) + })?; + Ok(Tree { + id, + kind, + scope, + root_id, + max_level: max_level.max(0) as u32, + status, + created_at: ms_to_utc(created_ms)?, + last_sealed_at: last_sealed_ms.map(ms_to_utc).transpose()?, + }) +} + +// ── Summary nodes ─────────────────────────────────────────────────────── + +/// Insert a sealed summary. Immutable — the caller must generate a fresh +/// id per seal. Idempotent on the primary key so retries of the same seal +/// transaction don't double-insert. +pub(crate) fn insert_summary_tx(tx: &Transaction<'_>, node: &SummaryNode) -> Result<()> { + tx.execute( + "INSERT OR IGNORE INTO mem_tree_summaries ( + id, tree_id, tree_kind, level, parent_id, + child_ids_json, content, token_count, + entities_json, topics_json, + time_range_start_ms, time_range_end_ms, + score, sealed_at_ms, deleted + ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15)", + params![ + node.id, + node.tree_id, + node.tree_kind.as_str(), + node.level, + node.parent_id, + serde_json::to_string(&node.child_ids)?, + node.content, + node.token_count, + serde_json::to_string(&node.entities)?, + serde_json::to_string(&node.topics)?, + node.time_range_start.timestamp_millis(), + node.time_range_end.timestamp_millis(), + node.score, + node.sealed_at.timestamp_millis(), + node.deleted as i64, + ], + ) + .with_context(|| format!("Failed to insert summary id={}", node.id))?; + Ok(()) +} + +/// Fetch one summary by id. Soft-deleted rows are returned with +/// `deleted = true` so callers can decide filtering policy. +pub fn get_summary(config: &Config, id: &str) -> Result> { + with_connection(config, |conn| { + let mut stmt = conn.prepare( + "SELECT id, tree_id, tree_kind, level, parent_id, + child_ids_json, content, token_count, + entities_json, topics_json, + time_range_start_ms, time_range_end_ms, + score, sealed_at_ms, deleted + FROM mem_tree_summaries WHERE id = ?1", + )?; + let row = stmt + .query_row(params![id], row_to_summary) + .optional() + .context("Failed to query summary by id")?; + Ok(row) + }) +} + +/// List sealed summaries for a tree at a given level, ordered by +/// `sealed_at` ascending. Skips tombstoned rows. +pub fn list_summaries_at_level( + config: &Config, + tree_id: &str, + level: u32, +) -> Result> { + with_connection(config, |conn| { + let mut stmt = conn.prepare( + "SELECT id, tree_id, tree_kind, level, parent_id, + child_ids_json, content, token_count, + entities_json, topics_json, + time_range_start_ms, time_range_end_ms, + score, sealed_at_ms, deleted + FROM mem_tree_summaries + WHERE tree_id = ?1 AND level = ?2 AND deleted = 0 + ORDER BY sealed_at_ms ASC", + )?; + let rows = stmt + .query_map(params![tree_id, level], row_to_summary)? + .collect::>>() + .context("Failed to collect summaries")?; + Ok(rows) + }) +} + +/// Count summaries in a tree (diagnostic helper). +pub fn count_summaries(config: &Config, tree_id: &str) -> Result { + with_connection(config, |conn| { + let n: i64 = conn + .query_row( + "SELECT COUNT(*) FROM mem_tree_summaries + WHERE tree_id = ?1 AND deleted = 0", + params![tree_id], + |r| r.get(0), + ) + .context("count summaries query")?; + Ok(n.max(0) as u64) + }) +} + +fn row_to_summary(row: &rusqlite::Row<'_>) -> rusqlite::Result { + let id: String = row.get(0)?; + let tree_id: String = row.get(1)?; + let tree_kind_s: String = row.get(2)?; + let level: i64 = row.get(3)?; + let parent_id: Option = row.get(4)?; + let child_ids_json: String = row.get(5)?; + let content: String = row.get(6)?; + let token_count: i64 = row.get(7)?; + let entities_json: String = row.get(8)?; + let topics_json: String = row.get(9)?; + let trs_ms: i64 = row.get(10)?; + let tre_ms: i64 = row.get(11)?; + let score: f64 = row.get(12)?; + let sealed_ms: i64 = row.get(13)?; + let deleted: i64 = row.get(14)?; + + let tree_kind = TreeKind::parse(&tree_kind_s).map_err(|e| { + rusqlite::Error::FromSqlConversionFailure(2, rusqlite::types::Type::Text, e.into()) + })?; + let child_ids: Vec = serde_json::from_str(&child_ids_json).map_err(|e| { + rusqlite::Error::FromSqlConversionFailure(5, rusqlite::types::Type::Text, Box::new(e)) + })?; + let entities: Vec = serde_json::from_str(&entities_json).map_err(|e| { + rusqlite::Error::FromSqlConversionFailure(8, rusqlite::types::Type::Text, Box::new(e)) + })?; + let topics: Vec = serde_json::from_str(&topics_json).map_err(|e| { + rusqlite::Error::FromSqlConversionFailure(9, rusqlite::types::Type::Text, Box::new(e)) + })?; + + Ok(SummaryNode { + id, + tree_id, + tree_kind, + level: level.max(0) as u32, + parent_id, + child_ids, + content, + token_count: token_count.max(0) as u32, + entities, + topics, + time_range_start: ms_to_utc(trs_ms)?, + time_range_end: ms_to_utc(tre_ms)?, + score: score as f32, + sealed_at: ms_to_utc(sealed_ms)?, + deleted: deleted != 0, + }) +} + +// ── Buffers ───────────────────────────────────────────────────────────── + +/// Read the current buffer at `(tree_id, level)` or return an empty one. +pub fn get_buffer(config: &Config, tree_id: &str, level: u32) -> Result { + with_connection(config, |conn| get_buffer_conn(conn, tree_id, level)) +} + +pub(crate) fn get_buffer_conn( + conn: &Connection, + tree_id: &str, + level: u32, +) -> Result { + let mut stmt = conn.prepare( + "SELECT tree_id, level, item_ids_json, token_sum, oldest_at_ms + FROM mem_tree_buffers WHERE tree_id = ?1 AND level = ?2", + )?; + let row = stmt + .query_row(params![tree_id, level], row_to_buffer) + .optional() + .context("Failed to query buffer")?; + Ok(row.unwrap_or_else(|| Buffer::empty(tree_id, level))) +} + +/// Upsert a buffer row. +pub(crate) fn upsert_buffer_tx(tx: &Transaction<'_>, buf: &Buffer) -> Result<()> { + let now_ms = Utc::now().timestamp_millis(); + tx.execute( + "INSERT INTO mem_tree_buffers ( + tree_id, level, item_ids_json, token_sum, oldest_at_ms, updated_at_ms + ) VALUES (?1, ?2, ?3, ?4, ?5, ?6) + ON CONFLICT(tree_id, level) DO UPDATE SET + item_ids_json = excluded.item_ids_json, + token_sum = excluded.token_sum, + oldest_at_ms = excluded.oldest_at_ms, + updated_at_ms = excluded.updated_at_ms", + params![ + buf.tree_id, + buf.level, + serde_json::to_string(&buf.item_ids)?, + buf.token_sum, + buf.oldest_at.map(|t| t.timestamp_millis()), + now_ms, + ], + ) + .with_context(|| { + format!( + "Failed to upsert buffer tree_id={} level={}", + buf.tree_id, buf.level + ) + })?; + Ok(()) +} + +/// Reset a buffer at `(tree_id, level)` to empty. Used at seal time: the +/// items move into a summary row and the buffer is cleared in the same tx. +pub(crate) fn clear_buffer_tx(tx: &Transaction<'_>, tree_id: &str, level: u32) -> Result<()> { + let empty = Buffer::empty(tree_id, level); + upsert_buffer_tx(tx, &empty) +} + +/// List all non-empty buffers ordered by `oldest_at_ms ASC`. Used by the +/// time-based flush pass. +pub fn list_stale_buffers( + config: &Config, + older_than: DateTime, +) -> Result> { + with_connection(config, |conn| { + let mut stmt = conn.prepare( + "SELECT tree_id, level, item_ids_json, token_sum, oldest_at_ms + FROM mem_tree_buffers + WHERE oldest_at_ms IS NOT NULL + AND oldest_at_ms <= ?1 + ORDER BY oldest_at_ms ASC", + )?; + let rows = stmt + .query_map(params![older_than.timestamp_millis()], row_to_buffer)? + .collect::>>() + .context("Failed to collect stale buffers")?; + Ok(rows) + }) +} + +fn row_to_buffer(row: &rusqlite::Row<'_>) -> rusqlite::Result { + let tree_id: String = row.get(0)?; + let level: i64 = row.get(1)?; + let item_ids_json: String = row.get(2)?; + let token_sum: i64 = row.get(3)?; + let oldest_ms: Option = row.get(4)?; + + let item_ids: Vec = serde_json::from_str(&item_ids_json).map_err(|e| { + rusqlite::Error::FromSqlConversionFailure(2, rusqlite::types::Type::Text, Box::new(e)) + })?; + let oldest_at = oldest_ms.map(ms_to_utc).transpose()?; + Ok(Buffer { + tree_id, + level: level.max(0) as u32, + item_ids, + token_sum, + oldest_at, + }) +} + +#[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) + } + + fn sample_tree(id: &str, scope: &str) -> Tree { + Tree { + id: id.to_string(), + kind: TreeKind::Source, + scope: scope.to_string(), + root_id: None, + max_level: 0, + status: TreeStatus::Active, + created_at: Utc.timestamp_millis_opt(1_700_000_000_000).unwrap(), + last_sealed_at: None, + } + } + + fn sample_summary(id: &str, tree_id: &str, level: u32) -> SummaryNode { + let ts = Utc.timestamp_millis_opt(1_700_000_000_000).unwrap(); + SummaryNode { + id: id.to_string(), + tree_id: tree_id.to_string(), + tree_kind: TreeKind::Source, + level, + parent_id: None, + child_ids: vec!["leaf-a".into(), "leaf-b".into()], + content: "seal content".into(), + token_count: 100, + entities: vec!["entity:alice".into()], + topics: vec!["#launch".into()], + time_range_start: ts, + time_range_end: ts, + score: 0.75, + sealed_at: ts, + deleted: false, + } + } + + #[test] + fn tree_round_trip() { + let (_tmp, cfg) = test_config(); + let t = sample_tree("tree-1", "slack:#eng"); + insert_tree(&cfg, &t).unwrap(); + let got = get_tree(&cfg, "tree-1").unwrap().unwrap(); + assert_eq!(got, t); + let by_scope = get_tree_by_scope(&cfg, TreeKind::Source, "slack:#eng") + .unwrap() + .unwrap(); + assert_eq!(by_scope.id, "tree-1"); + } + + #[test] + fn duplicate_scope_fails() { + let (_tmp, cfg) = test_config(); + insert_tree(&cfg, &sample_tree("t1", "slack:#eng")).unwrap(); + let dup = sample_tree("t2", "slack:#eng"); + assert!(insert_tree(&cfg, &dup).is_err()); + } + + #[test] + fn summary_insert_and_fetch() { + let (_tmp, cfg) = test_config(); + insert_tree(&cfg, &sample_tree("tree-1", "slack:#eng")).unwrap(); + let node = sample_summary("sum-1", "tree-1", 1); + with_connection(&cfg, |conn| { + let tx = conn.unchecked_transaction()?; + insert_summary_tx(&tx, &node)?; + tx.commit()?; + Ok(()) + }) + .unwrap(); + let got = get_summary(&cfg, "sum-1").unwrap().unwrap(); + assert_eq!(got, node); + let at_level = list_summaries_at_level(&cfg, "tree-1", 1).unwrap(); + assert_eq!(at_level.len(), 1); + assert_eq!(count_summaries(&cfg, "tree-1").unwrap(), 1); + } + + #[test] + fn summary_insert_is_idempotent_on_id() { + let (_tmp, cfg) = test_config(); + insert_tree(&cfg, &sample_tree("tree-1", "slack:#eng")).unwrap(); + let node = sample_summary("sum-1", "tree-1", 1); + with_connection(&cfg, |conn| { + let tx = conn.unchecked_transaction()?; + insert_summary_tx(&tx, &node)?; + insert_summary_tx(&tx, &node)?; + tx.commit()?; + Ok(()) + }) + .unwrap(); + assert_eq!(count_summaries(&cfg, "tree-1").unwrap(), 1); + } + + #[test] + fn buffer_upsert_and_clear() { + let (_tmp, cfg) = test_config(); + insert_tree(&cfg, &sample_tree("tree-1", "slack:#eng")).unwrap(); + let ts = Utc.timestamp_millis_opt(1_700_000_000_000).unwrap(); + let buf = Buffer { + tree_id: "tree-1".into(), + level: 0, + item_ids: vec!["leaf-a".into(), "leaf-b".into()], + token_sum: 500, + oldest_at: Some(ts), + }; + with_connection(&cfg, |conn| { + let tx = conn.unchecked_transaction()?; + upsert_buffer_tx(&tx, &buf)?; + tx.commit()?; + Ok(()) + }) + .unwrap(); + let got = get_buffer(&cfg, "tree-1", 0).unwrap(); + assert_eq!(got, buf); + + with_connection(&cfg, |conn| { + let tx = conn.unchecked_transaction()?; + clear_buffer_tx(&tx, "tree-1", 0)?; + tx.commit()?; + Ok(()) + }) + .unwrap(); + let cleared = get_buffer(&cfg, "tree-1", 0).unwrap(); + assert!(cleared.is_empty()); + assert_eq!(cleared.token_sum, 0); + assert!(cleared.oldest_at.is_none()); + } + + #[test] + fn get_buffer_returns_empty_when_missing() { + let (_tmp, cfg) = test_config(); + insert_tree(&cfg, &sample_tree("tree-1", "slack:#eng")).unwrap(); + let got = get_buffer(&cfg, "tree-1", 0).unwrap(); + assert!(got.is_empty()); + assert_eq!(got.tree_id, "tree-1"); + } + + #[test] + fn update_tree_after_seal_persists() { + let (_tmp, cfg) = test_config(); + insert_tree(&cfg, &sample_tree("tree-1", "slack:#eng")).unwrap(); + let sealed_at = Utc.timestamp_millis_opt(1_700_000_123_000).unwrap(); + with_connection(&cfg, |conn| { + let tx = conn.unchecked_transaction()?; + update_tree_after_seal_tx(&tx, "tree-1", "sum-1", 1, sealed_at)?; + tx.commit()?; + Ok(()) + }) + .unwrap(); + let got = get_tree(&cfg, "tree-1").unwrap().unwrap(); + assert_eq!(got.root_id.as_deref(), Some("sum-1")); + assert_eq!(got.max_level, 1); + assert_eq!(got.last_sealed_at, Some(sealed_at)); + } + + #[test] + fn list_stale_buffers_orders_by_age() { + let (_tmp, cfg) = test_config(); + insert_tree(&cfg, &sample_tree("tree-1", "slack:#eng")).unwrap(); + let t0 = Utc.timestamp_millis_opt(1_700_000_000_000).unwrap(); + let t1 = Utc.timestamp_millis_opt(1_700_000_010_000).unwrap(); + let t2 = Utc.timestamp_millis_opt(1_700_000_020_000).unwrap(); + with_connection(&cfg, |conn| { + let tx = conn.unchecked_transaction()?; + upsert_buffer_tx( + &tx, + &Buffer { + tree_id: "tree-1".into(), + level: 0, + item_ids: vec!["a".into()], + token_sum: 10, + oldest_at: Some(t0), + }, + )?; + upsert_buffer_tx( + &tx, + &Buffer { + tree_id: "tree-1".into(), + level: 1, + item_ids: vec!["b".into()], + token_sum: 20, + oldest_at: Some(t1), + }, + )?; + tx.commit()?; + Ok(()) + }) + .unwrap(); + let stale = list_stale_buffers(&cfg, t2).unwrap(); + assert_eq!(stale.len(), 2); + assert_eq!(stale[0].level, 0); + assert_eq!(stale[1].level, 1); + // Filter out the first: only level-1 should come back. + let only_later = list_stale_buffers(&cfg, t0).unwrap(); + assert_eq!(only_later.len(), 1); + assert_eq!(only_later[0].level, 0); + } +} diff --git a/src/openhuman/memory/tree/source_tree/summariser/inert.rs b/src/openhuman/memory/tree/source_tree/summariser/inert.rs new file mode 100644 index 000000000..e1f5584af --- /dev/null +++ b/src/openhuman/memory/tree/source_tree/summariser/inert.rs @@ -0,0 +1,190 @@ +//! Deterministic fallback summariser (#709). +//! +//! `InertSummariser` concatenates each input's content, separated by a +//! blank line, and hard-truncates to `ctx.token_budget`. It also unions +//! the entity and topic sets (dedup while preserving first-seen order). +//! The goal is not fidelity — it's a stable, dependency-free baseline so +//! tree mechanics (sealing, cascade, roots) can be tested without an LLM. + +use anyhow::Result; +use async_trait::async_trait; + +use crate::openhuman::memory::tree::source_tree::summariser::{ + SummaryContext, SummaryInput, SummaryOutput, Summariser, +}; +use crate::openhuman::memory::tree::types::approx_token_count; + +/// Default prefix applied to each contribution in the joined body. Keeps +/// provenance visible to a human reading the raw summary. +const PROVENANCE_PREFIX: &str = "— "; + +pub struct InertSummariser; + +impl InertSummariser { + pub fn new() -> Self { + Self + } +} + +impl Default for InertSummariser { + fn default() -> Self { + Self::new() + } +} + +#[async_trait] +impl Summariser for InertSummariser { + async fn summarise( + &self, + inputs: &[SummaryInput], + ctx: &SummaryContext<'_>, + ) -> Result { + let mut parts: Vec = Vec::with_capacity(inputs.len()); + for inp in inputs { + let trimmed = inp.content.trim(); + if trimmed.is_empty() { + continue; + } + parts.push(format!("{}{}", PROVENANCE_PREFIX, trimmed)); + } + let joined = parts.join("\n\n"); + + let (content, token_count) = truncate_to_budget(&joined, ctx.token_budget); + + let entities = union_preserve_order(inputs.iter().map(|i| i.entities.as_slice())); + let topics = union_preserve_order(inputs.iter().map(|i| i.topics.as_slice())); + + log::debug!( + "[source_tree::summariser::inert] sealed tree_id={} level={} inputs={} tokens={}", + ctx.tree_id, + ctx.target_level, + inputs.len(), + token_count + ); + + Ok(SummaryOutput { + content, + token_count, + entities, + topics, + }) + } +} + +/// Truncate `text` to fit within `budget` approximate tokens. Returns the +/// (possibly truncated) body and its recomputed token count. Truncation is +/// done on character boundaries — `approx_token_count` assumes ~4 chars +/// per token so we clamp character length to `budget * 4`. +fn truncate_to_budget(text: &str, budget: u32) -> (String, u32) { + let initial = approx_token_count(text); + if initial <= budget { + return (text.to_string(), initial); + } + // Character ceiling derived from the same ~4 chars/token heuristic. + let char_ceiling = (budget as usize).saturating_mul(4); + let truncated: String = text.chars().take(char_ceiling).collect(); + let tokens = approx_token_count(&truncated); + (truncated, tokens) +} + +fn union_preserve_order<'a, I>(iter: I) -> Vec +where + I: IntoIterator, +{ + let mut out: Vec = Vec::new(); + for group in iter { + for item in group { + if !out.iter().any(|existing| existing == item) { + out.push(item.clone()); + } + } + } + out +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::openhuman::memory::tree::source_tree::types::TreeKind; + use chrono::Utc; + + fn sample_input(id: &str, content: &str, entities: &[&str]) -> SummaryInput { + let ts = Utc::now(); + SummaryInput { + id: id.to_string(), + content: content.to_string(), + token_count: approx_token_count(content), + entities: entities.iter().map(|s| s.to_string()).collect(), + topics: Vec::new(), + time_range_start: ts, + time_range_end: ts, + score: 0.5, + } + } + + fn test_ctx() -> SummaryContext<'static> { + SummaryContext { + tree_id: "tree-1", + tree_kind: TreeKind::Source, + target_level: 1, + token_budget: 10_000, + } + } + + #[tokio::test] + async fn concats_inputs_with_provenance_prefix() { + let s = InertSummariser::default(); + let inputs = vec![ + sample_input("a", "hello world", &[]), + sample_input("b", "second contribution", &[]), + ]; + let out = s.summarise(&inputs, &test_ctx()).await.unwrap(); + assert!(out.content.contains(PROVENANCE_PREFIX)); + assert!(out.content.contains("hello world")); + assert!(out.content.contains("second contribution")); + assert_eq!(out.token_count, approx_token_count(&out.content)); + } + + #[tokio::test] + async fn unions_entities_preserving_order_and_dedupe() { + let s = InertSummariser::default(); + let inputs = vec![ + sample_input("a", "x", &["entity:alice", "entity:bob"]), + sample_input("b", "y", &["entity:bob", "entity:carol"]), + ]; + let out = s.summarise(&inputs, &test_ctx()).await.unwrap(); + assert_eq!( + out.entities, + vec![ + "entity:alice".to_string(), + "entity:bob".to_string(), + "entity:carol".to_string(), + ] + ); + } + + #[tokio::test] + async fn truncates_when_over_budget() { + let s = InertSummariser::default(); + let long_text = "a".repeat(100); + let inputs = vec![sample_input("a", &long_text, &[])]; + let mut ctx = test_ctx(); + ctx.token_budget = 5; // way under — should truncate hard + let out = s.summarise(&inputs, &ctx).await.unwrap(); + assert!(out.token_count <= ctx.token_budget + 1); + assert!(out.content.len() < long_text.len() + PROVENANCE_PREFIX.len()); + } + + #[tokio::test] + async fn skips_empty_contributions() { + let s = InertSummariser::default(); + let inputs = vec![ + sample_input("a", " ", &[]), + sample_input("b", "kept", &[]), + ]; + let out = s.summarise(&inputs, &test_ctx()).await.unwrap(); + assert!(out.content.contains("kept")); + // exactly one provenance prefix should appear + assert_eq!(out.content.matches(PROVENANCE_PREFIX).count(), 1); + } +} diff --git a/src/openhuman/memory/tree/source_tree/summariser/mod.rs b/src/openhuman/memory/tree/source_tree/summariser/mod.rs new file mode 100644 index 000000000..a273643b2 --- /dev/null +++ b/src/openhuman/memory/tree/source_tree/summariser/mod.rs @@ -0,0 +1,62 @@ +//! Summariser trait + fallback (#709). +//! +//! A summariser folds N buffered items into one sealed summary. Phase 3a +//! ships an `InertSummariser` that concatenates the contributions and +//! truncates to the token budget — enough to make the tree mechanics +//! observable end-to-end without requiring an LLM. Real summarisation +//! (Ollama, etc.) can slot in by implementing the trait. + +use anyhow::Result; +use async_trait::async_trait; +use chrono::{DateTime, Utc}; + +use crate::openhuman::memory::tree::source_tree::types::TreeKind; + +pub mod inert; + +/// One contribution being folded — either a raw leaf (chunk) at L0→L1, or +/// a lower-level summary at L_n→L_{n+1}. +#[derive(Clone, Debug)] +pub struct SummaryInput { + /// Primary key of the contribution (chunk id or summary id). + pub id: String, + pub content: String, + pub token_count: u32, + pub entities: Vec, + pub topics: Vec, + pub time_range_start: DateTime, + pub time_range_end: DateTime, + /// Score signal from scoring (for leaves) or parent seal (for summaries). + pub score: f32, +} + +/// Opaque context passed to the summariser — lets implementations log / +/// identify which tree is being sealed without threading config globally. +#[derive(Clone, Debug)] +pub struct SummaryContext<'a> { + pub tree_id: &'a str, + pub tree_kind: TreeKind, + pub target_level: u32, + pub token_budget: u32, +} + +/// Output of a summariser invocation. +#[derive(Clone, Debug)] +pub struct SummaryOutput { + pub content: String, + pub token_count: u32, + pub entities: Vec, + pub topics: Vec, +} + +#[async_trait] +pub trait Summariser: Send + Sync { + /// Fold the inputs into a single summary. `ctx.token_budget` is an + /// upper bound on the produced `token_count`; implementations SHOULD + /// stay well under it so parents have room to include this summary. + async fn summarise( + &self, + inputs: &[SummaryInput], + ctx: &SummaryContext<'_>, + ) -> Result; +} diff --git a/src/openhuman/memory/tree/source_tree/types.rs b/src/openhuman/memory/tree/source_tree/types.rs new file mode 100644 index 000000000..2cd010e3c --- /dev/null +++ b/src/openhuman/memory/tree/source_tree/types.rs @@ -0,0 +1,215 @@ +//! Core types for Phase 3a — summary trees, per-source bucket-seal (#709). +//! +//! These types sit on top of Phase 1's chunk leaves. A [`Tree`] groups leaves +//! under one scope (e.g. one chat channel, one email account). When a +//! [`Buffer`] at some level accumulates enough tokens, its contents seal +//! into a [`SummaryNode`] at level+1 and the buffer clears. Summary nodes +//! are immutable once emitted — updates to children use the Phase 1/2 +//! tombstone pattern, never rewrite parents. + +use chrono::{DateTime, Utc}; +use serde::{Deserialize, Serialize}; + +/// What kind of tree this is. Source trees live per ingest source; topic +/// and global trees are introduced in Phase 3b/3c and share the same +/// schema. +#[derive(Clone, Copy, Debug, Eq, PartialEq, Hash, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +#[non_exhaustive] +pub enum TreeKind { + /// One tree per ingest source (e.g. `chat:slack:#eng`, `email:gmail:user`). + Source, + /// Reserved for Phase 3c — per-entity/topic tree. + Topic, + /// Reserved for Phase 3b — cross-source daily digest tree. + Global, +} + +impl TreeKind { + pub fn as_str(self) -> &'static str { + match self { + Self::Source => "source", + Self::Topic => "topic", + Self::Global => "global", + } + } + + pub fn parse(s: &str) -> Result { + match s { + "source" => Ok(Self::Source), + "topic" => Ok(Self::Topic), + "global" => Ok(Self::Global), + other => Err(format!("unknown tree kind: {other}")), + } + } +} + +/// Activity state of a tree. Archived trees stay queryable but don't accept +/// new leaves — used by Phase 3c when a topic tree's entity goes cold. +#[derive(Clone, Copy, Debug, Eq, PartialEq, Hash, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum TreeStatus { + Active, + Archived, +} + +impl TreeStatus { + pub fn as_str(self) -> &'static str { + match self { + Self::Active => "active", + Self::Archived => "archived", + } + } + + pub fn parse(s: &str) -> Result { + match s { + "active" => Ok(Self::Active), + "archived" => Ok(Self::Archived), + other => Err(format!("unknown tree status: {other}")), + } + } +} + +/// One summary-tree instance. +/// +/// `root_id` is `None` until the first seal emits an L1 node. `max_level` +/// tracks the highest level that has ever sealed; `root_id` points at the +/// current top node at that level (changes on root-split). +#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)] +pub struct Tree { + pub id: String, + pub kind: TreeKind, + /// Logical identifier for what the tree covers. Format conventions: + /// - Source: `::` or the chunk's + /// `source_id` directly (Phase 3a uses the chunk source_id verbatim) + /// - Topic: canonical entity id + /// - Global: the literal string `"global"` + pub scope: String, + pub root_id: Option, + pub max_level: u32, + pub status: TreeStatus, + pub created_at: DateTime, + pub last_sealed_at: Option>, +} + +/// A sealed summary node — one level above raw leaves. +/// +/// `child_ids` points at the concrete children that were in the buffer when +/// this node sealed. For L1 nodes those are leaf `chunk.id`s; for L2+ they +/// are lower-level summary ids. Relation is fixed at seal time — never +/// modified afterwards. +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] +pub struct SummaryNode { + pub id: String, + pub tree_id: String, + pub tree_kind: TreeKind, + /// 1 for summaries over raw leaves, 2 over L1 summaries, and so on. + pub level: u32, + pub parent_id: Option, + pub child_ids: Vec, + /// Summariser output. Typical target: 800–1500 tokens. + pub content: String, + pub token_count: u32, + /// Curated subset of children's entity canonical-ids. + pub entities: Vec, + /// Curated topic labels (hashtag-like short phrases). + pub topics: Vec, + pub time_range_start: DateTime, + pub time_range_end: DateTime, + /// Max of children's scores at seal time — cheap heuristic, preserved + /// for reranking in Phase 4. + pub score: f32, + pub sealed_at: DateTime, + /// Tombstone flag — stays `false` in Phase 3a since summaries are + /// immutable. Reserved for future cleanup passes (e.g. archive cascade). + pub deleted: bool, +} + +/// Unsealed frontier at a given `(tree_id, level)`. One row per level per +/// tree. `oldest_at` is `None` when the buffer is empty; used by the +/// time-based flush trigger. +#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)] +pub struct Buffer { + pub tree_id: String, + pub level: u32, + pub item_ids: Vec, + pub token_sum: i64, + pub oldest_at: Option>, +} + +impl Buffer { + /// Empty buffer at the given key. + pub fn empty(tree_id: &str, level: u32) -> Self { + Self { + tree_id: tree_id.to_string(), + level, + item_ids: Vec::new(), + token_sum: 0, + oldest_at: None, + } + } + + pub fn is_empty(&self) -> bool { + self.item_ids.is_empty() + } + + /// Whether the buffer's oldest item is older than `max_age`. Returns + /// `false` for an empty buffer. + pub fn is_stale(&self, now: DateTime, max_age: chrono::Duration) -> bool { + match self.oldest_at { + Some(ts) => now.signed_duration_since(ts) > max_age, + None => false, + } + } +} + +/// Token ceiling for one summariser invocation — aligned with the Phase 1 +/// chunker ceiling so a single leaf never busts a seal on its own. +pub const TOKEN_BUDGET: u32 = 10_000; + +/// Default age at which a non-empty buffer is force-sealed even under the +/// token budget. Keeps recent activity from stalling waiting for more +/// leaves that may never arrive. +pub const DEFAULT_FLUSH_AGE_SECS: i64 = 7 * 24 * 60 * 60; + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn tree_kind_round_trip() { + for k in [TreeKind::Source, TreeKind::Topic, TreeKind::Global] { + assert_eq!(TreeKind::parse(k.as_str()).unwrap(), k); + } + assert!(TreeKind::parse("bogus").is_err()); + } + + #[test] + fn tree_status_round_trip() { + for s in [TreeStatus::Active, TreeStatus::Archived] { + assert_eq!(TreeStatus::parse(s.as_str()).unwrap(), s); + } + assert!(TreeStatus::parse("live").is_err()); + } + + #[test] + fn empty_buffer_is_not_stale() { + let b = Buffer::empty("t1", 0); + assert!(b.is_empty()); + assert!(!b.is_stale(Utc::now(), chrono::Duration::zero())); + } + + #[test] + fn stale_buffer_detected() { + let past = Utc::now() - chrono::Duration::hours(10); + let b = Buffer { + tree_id: "t1".into(), + level: 0, + item_ids: vec!["leaf-1".into()], + token_sum: 100, + oldest_at: Some(past), + }; + assert!(b.is_stale(Utc::now(), chrono::Duration::hours(1))); + assert!(!b.is_stale(Utc::now(), chrono::Duration::hours(20))); + } +} diff --git a/src/openhuman/memory/tree/store.rs b/src/openhuman/memory/tree/store.rs index 5c686dcc6..53e7fd7d8 100644 --- a/src/openhuman/memory/tree/store.rs +++ b/src/openhuman/memory/tree/store.rs @@ -88,6 +88,71 @@ CREATE INDEX IF NOT EXISTS idx_mem_tree_entity_index_node ON mem_tree_entity_index(node_id); CREATE INDEX IF NOT EXISTS idx_mem_tree_entity_index_timestamp ON mem_tree_entity_index(timestamp_ms); + +-- Phase 3a (#709): summary trees / bucket-seal. +-- `mem_tree_trees` tracks one tree per scope (source/topic/global). +CREATE TABLE IF NOT EXISTS mem_tree_trees ( + id TEXT PRIMARY KEY, + kind TEXT NOT NULL, + scope TEXT NOT NULL, + root_id TEXT, + max_level INTEGER NOT NULL DEFAULT 0, + status TEXT NOT NULL DEFAULT 'active', + created_at_ms INTEGER NOT NULL, + last_sealed_at_ms INTEGER +); + +CREATE UNIQUE INDEX IF NOT EXISTS idx_mem_tree_trees_kind_scope + ON mem_tree_trees(kind, scope); +CREATE INDEX IF NOT EXISTS idx_mem_tree_trees_status + ON mem_tree_trees(status); + +-- `mem_tree_summaries` holds sealed summary nodes. Immutable once written +-- (Phase 3a). `deleted` is reserved for future archive cascades. +CREATE TABLE IF NOT EXISTS mem_tree_summaries ( + id TEXT PRIMARY KEY, + tree_id TEXT NOT NULL, + tree_kind TEXT NOT NULL, + level INTEGER NOT NULL, + parent_id TEXT, + child_ids_json TEXT NOT NULL DEFAULT '[]', + content TEXT NOT NULL, + token_count INTEGER NOT NULL, + entities_json TEXT NOT NULL DEFAULT '[]', + topics_json TEXT NOT NULL DEFAULT '[]', + time_range_start_ms INTEGER NOT NULL, + time_range_end_ms INTEGER NOT NULL, + score REAL NOT NULL DEFAULT 0.0, + sealed_at_ms INTEGER NOT NULL, + deleted INTEGER NOT NULL DEFAULT 0, + FOREIGN KEY (tree_id) REFERENCES mem_tree_trees(id) +); + +CREATE INDEX IF NOT EXISTS idx_mem_tree_summaries_tree_level + ON mem_tree_summaries(tree_id, level); +CREATE INDEX IF NOT EXISTS idx_mem_tree_summaries_parent + ON mem_tree_summaries(parent_id); +CREATE INDEX IF NOT EXISTS idx_mem_tree_summaries_sealed_at + ON mem_tree_summaries(sealed_at_ms); +CREATE INDEX IF NOT EXISTS idx_mem_tree_summaries_deleted + ON mem_tree_summaries(deleted); + +-- `mem_tree_buffers` holds the unsealed frontier per (tree, level). One row +-- per active level per tree; deleted when the buffer seals (clears) in the +-- same transaction as the new summary node row. +CREATE TABLE IF NOT EXISTS mem_tree_buffers ( + tree_id TEXT NOT NULL, + level INTEGER NOT NULL, + item_ids_json TEXT NOT NULL DEFAULT '[]', + token_sum INTEGER NOT NULL DEFAULT 0, + oldest_at_ms INTEGER, + updated_at_ms INTEGER NOT NULL, + PRIMARY KEY (tree_id, level), + FOREIGN KEY (tree_id) REFERENCES mem_tree_trees(id) +); + +CREATE INDEX IF NOT EXISTS idx_mem_tree_buffers_oldest + ON mem_tree_buffers(oldest_at_ms); "; /// Upsert a batch of chunks atomically. @@ -355,6 +420,10 @@ pub(crate) fn with_connection( // "no LLM signal available" by readers. add_column_if_missing(&conn, "mem_tree_score", "llm_importance", "REAL")?; add_column_if_missing(&conn, "mem_tree_score", "llm_importance_reason", "TEXT")?; + // Phase 3a (#709): parent-summary backlink on leaves. Populated when + // the L0 buffer seals into an L1 summary so traversal can walk + // leaf → parent without scanning `mem_tree_summaries.child_ids_json`. + add_column_if_missing(&conn, "mem_tree_chunks", "parent_summary_id", "TEXT")?; f(&conn) }