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/score/extract/types.rs b/src/openhuman/memory/tree/score/extract/types.rs index ca737c50d..8e91c8437 100644 --- a/src/openhuman/memory/tree/score/extract/types.rs +++ b/src/openhuman/memory/tree/score/extract/types.rs @@ -33,6 +33,13 @@ pub enum EntityKind { Event, Product, Misc, + // Thematic — scorer-surfaced topics (hashtag-like short phrases or + // LLM-extracted themes). Promoted into the canonical entity stream + // by the resolver so Phase 3c topic trees can route on themes the + // same way they route on people/orgs. A chunk saying "Phoenix + // migration ships Friday" emits `topic:phoenix` and `topic:migration` + // in addition to any emails/hashtags the mechanical extractors find. + Topic, } impl EntityKind { @@ -48,6 +55,7 @@ impl EntityKind { Self::Event => "event", Self::Product => "product", Self::Misc => "misc", + Self::Topic => "topic", } } @@ -63,6 +71,7 @@ impl EntityKind { "event" => Ok(Self::Event), "product" => Ok(Self::Product), "misc" => Ok(Self::Misc), + "topic" => Ok(Self::Topic), other => Err(format!("unknown entity kind: {other}")), } } diff --git a/src/openhuman/memory/tree/score/resolver.rs b/src/openhuman/memory/tree/score/resolver.rs index 14cc56faf..cc5201b7b 100644 --- a/src/openhuman/memory/tree/score/resolver.rs +++ b/src/openhuman/memory/tree/score/resolver.rs @@ -28,8 +28,15 @@ pub struct CanonicalEntity { /// Same surface form (after normalisation) → same `canonical_id` regardless /// of how many times it appears in a chunk. Preserves source spans by /// emitting one [`CanonicalEntity`] per occurrence. +/// +/// Extracted **topics** are also promoted into the canonical stream under +/// [`EntityKind::Topic`] so downstream routing (Phase 3c topic trees) can +/// treat themes as first-class scope alongside people/orgs. Topics have no +/// source span (they're derived from the whole chunk, not a specific +/// substring), so `span_start` / `span_end` are both `0` for topic rows — +/// readers should key on `kind` instead of span when span-awareness matters. pub fn canonicalise(extracted: &ExtractedEntities) -> Vec { - extracted + let mut out: Vec = extracted .entities .iter() .map(|e| CanonicalEntity { @@ -40,7 +47,35 @@ pub fn canonicalise(extracted: &ExtractedEntities) -> Vec { span_end: e.span_end, score: e.score, }) - .collect() + .collect(); + + // Promote topics. Dedup against the entities we already emitted so a + // hashtag like `#launch` and a topic label `"launch"` don't both land + // as the same canonical id with the same kind — the hashtag keeps its + // Hashtag kind, the topic gets Topic kind, and `canonical_id_for` + // makes them distinguishable: `hashtag:launch` vs `topic:launch`. + for topic in &extracted.topics { + let canonical_id = canonical_id_for(EntityKind::Topic, &topic.label); + // Dedup within the topic set in case the scorer produces the same + // label twice (LLM + regex overlap). Entities under other kinds + // aren't dedup targets — `topic:launch` and `hashtag:launch` are + // intentionally separate. + if out + .iter() + .any(|e| e.kind == EntityKind::Topic && e.canonical_id == canonical_id) + { + continue; + } + out.push(CanonicalEntity { + canonical_id, + kind: EntityKind::Topic, + surface: topic.label.clone(), + span_start: 0, + span_end: 0, + score: topic.score, + }); + } + out } /// Canonical id form per kind. Deterministic so the same surface always @@ -135,4 +170,96 @@ mod tests { canonical_id_for(EntityKind::Person, "alice") ); } + + // ── Topic canonicalisation (#709 / Phase 3c topic-tree scope) ──── + + use crate::openhuman::memory::tree::score::extract::ExtractedTopic; + + fn topic(label: &str, score: f32) -> ExtractedTopic { + ExtractedTopic { + label: label.to_string(), + score, + } + } + + #[test] + fn topics_are_promoted_to_canonical_entities() { + let ex = ExtractedEntities { + entities: vec![], + topics: vec![topic("phoenix", 0.72), topic("migration", 0.60)], + llm_importance: None, + llm_importance_reason: None, + }; + let out = canonicalise(&ex); + assert_eq!(out.len(), 2); + assert_eq!(out[0].kind, EntityKind::Topic); + assert_eq!(out[0].canonical_id, "topic:phoenix"); + assert!((out[0].score - 0.72).abs() < 1e-6); + assert_eq!(out[1].canonical_id, "topic:migration"); + } + + #[test] + fn topic_canonicalisation_lowercases() { + let ex = ExtractedEntities { + entities: vec![], + topics: vec![topic("Phoenix", 1.0), topic("PHOENIX", 0.5)], + llm_importance: None, + llm_importance_reason: None, + }; + let out = canonicalise(&ex); + // Both normalise to "topic:phoenix" — second occurrence is deduped. + assert_eq!(out.len(), 1); + assert_eq!(out[0].canonical_id, "topic:phoenix"); + // First-seen surface is preserved. + assert_eq!(out[0].surface, "Phoenix"); + } + + #[test] + fn hashtag_and_topic_with_same_label_coexist() { + // "#launch" regex → EntityKind::Hashtag, LLM theme "launch" → Topic. + // They stay as two distinct canonical entities — different kind, + // different canonical_id prefix. + let ex = ExtractedEntities { + entities: vec![ExtractedEntity { + kind: EntityKind::Hashtag, + text: "launch".into(), + span_start: 0, + span_end: 6, + score: 1.0, + }], + topics: vec![topic("launch", 0.8)], + llm_importance: None, + llm_importance_reason: None, + }; + let out = canonicalise(&ex); + assert_eq!(out.len(), 2); + assert_eq!(out[0].kind, EntityKind::Hashtag); + assert_eq!(out[0].canonical_id, "hashtag:launch"); + assert_eq!(out[1].kind, EntityKind::Topic); + assert_eq!(out[1].canonical_id, "topic:launch"); + } + + #[test] + fn canonicalise_mixes_entities_and_topics_in_order() { + // Entities come first, topics appended after — downstream callers + // (e.g. routing) can rely on this ordering if they ever need it. + let ex = ExtractedEntities { + entities: vec![entity(EntityKind::Email, "alice@example.com")], + topics: vec![topic("phoenix", 0.7)], + llm_importance: None, + llm_importance_reason: None, + }; + let out = canonicalise(&ex); + assert_eq!(out.len(), 2); + assert_eq!(out[0].kind, EntityKind::Email); + assert_eq!(out[1].kind, EntityKind::Topic); + } + + #[test] + fn topic_entity_kind_round_trips_through_parse() { + // Defence in depth: ensure the new Topic variant survives the + // round-trip used by mem_tree_entity_index on read. + assert_eq!(EntityKind::parse("topic"), Ok(EntityKind::Topic)); + assert_eq!(EntityKind::Topic.as_str(), "topic"); + } } diff --git a/src/openhuman/memory/tree/score/store.rs b/src/openhuman/memory/tree/score/store.rs index bed44b455..ce9eeee31 100644 --- a/src/openhuman/memory/tree/score/store.rs +++ b/src/openhuman/memory/tree/score/store.rs @@ -223,6 +223,65 @@ pub(crate) fn clear_entity_index_for_node_tx(tx: &Transaction<'_>, node_id: &str Ok(n) } +/// Index summary-node entities by canonical id only. Summary-level entity +/// metadata is LLM-derived (Phase 3a #709) — the summariser emits a +/// curated list of canonical ids without per-occurrence span/surface data. +/// +/// Writes the kind prefix (everything before the first `:`) into the +/// `entity_kind` column so [`lookup_entity`]'s `EntityKind::parse()` keeps +/// round-tripping on summary rows. `surface` stores the full canonical id +/// as a stable placeholder — at the summary level we have no per-occurrence +/// span to recover, and the id is always unique. The summary's score is +/// reused for each of its entities. +/// +/// Callers should prefer the regular [`index_entities_tx`] for leaves, +/// where span/surface are meaningful. +pub(crate) fn index_summary_entity_ids_tx( + tx: &Transaction<'_>, + entity_ids: &[String], + node_id: &str, + score: f32, + timestamp_ms: i64, + tree_id: Option<&str>, +) -> Result { + if entity_ids.is_empty() { + return Ok(0); + } + let mut stmt = tx.prepare( + "INSERT OR REPLACE INTO mem_tree_entity_index ( + entity_id, node_id, node_kind, entity_kind, surface, + score, timestamp_ms, tree_id + ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)", + )?; + for canonical_id in entity_ids { + // Canonical ids follow Phase 2's ":" convention. + // Without this split, `entity_kind` would hold the full id and + // `lookup_entity`'s `EntityKind::parse()` would fail at read time, + // poisoning any mixed leaf/summary lookup. + let entity_kind = match canonical_id.split_once(':') { + Some((kind, _)) => kind, + None => { + log::warn!( + "[memory_tree::score::store] summary entity id missing ':' — \ + storing as-is: {canonical_id}" + ); + canonical_id.as_str() + } + }; + stmt.execute(params![ + canonical_id, + node_id, + "summary", + entity_kind, + canonical_id, + score, + timestamp_ms, + tree_id, + ])?; + } + Ok(entity_ids.len()) +} + pub(crate) fn index_entities_tx( tx: &Transaction<'_>, entities: &[CanonicalEntity], @@ -485,4 +544,58 @@ mod tests { let hits = lookup_entity(&cfg, "email:alice", Some(2)).unwrap(); assert_eq!(hits.len(), 2); } + + /// Regression: `index_summary_entity_ids_tx` must write a parseable + /// `entity_kind` (the "" prefix before `:`) so `lookup_entity` + /// can still round-trip rows through `EntityKind::parse`. Earlier code + /// stored the full canonical id, which poisoned lookups mixing leaf + /// and summary hits. See PR #789 CodeRabbit review. + #[test] + fn summary_entity_index_kind_is_parseable() { + use crate::openhuman::memory::tree::store::with_connection; + + let (_tmp, cfg) = test_config(); + + // Seed a leaf hit so lookup_entity has something leafy to mix + // with the summary hit — this reproduces the mixed-row crash. + let leaf_entity = sample_entity("alice"); + index_entity(&cfg, &leaf_entity, "leaf-1", "leaf", 1000, Some("tree-1")).unwrap(); + + // Write a summary row via the tx helper under test. + with_connection(&cfg, |conn| { + let tx = conn.unchecked_transaction()?; + let n = index_summary_entity_ids_tx( + &tx, + &["email:alice@example.com".into(), "hashtag:launch-q2".into()], + "summary-1", + 0.84, + 2000, + Some("tree-1"), + )?; + assert_eq!(n, 2); + tx.commit()?; + Ok(()) + }) + .unwrap(); + + // Before the fix: lookup_entity would fail on the summary row + // because entity_kind was "email:alice@example.com" and + // EntityKind::parse rejects it. After the fix, the column stores + // "email" and the lookup succeeds with both rows. + let hits = lookup_entity(&cfg, "email:alice@example.com", None).unwrap(); + assert_eq!(hits.len(), 1, "summary row should be discoverable"); + assert_eq!(hits[0].node_id, "summary-1"); + assert_eq!(hits[0].node_kind, "summary"); + assert_eq!(hits[0].entity_kind, EntityKind::Email); + + // Hashtag row parses as its own kind too. + let hits = lookup_entity(&cfg, "hashtag:launch-q2", None).unwrap(); + assert_eq!(hits.len(), 1); + assert_eq!(hits[0].entity_kind, EntityKind::Hashtag); + + // Mixing leaf + summary entity ids in one lookup also parses cleanly. + let hits = lookup_entity(&cfg, "email:alice", None).unwrap(); + assert_eq!(hits.len(), 1); + assert_eq!(hits[0].entity_kind, EntityKind::Email); + } } 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..51316cd91 --- /dev/null +++ b/src/openhuman/memory/tree/source_tree/bucket_seal.rs @@ -0,0 +1,566 @@ +//! 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::{ + Summariser, SummaryContext, SummaryInput, +}; +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)?; + // Idempotent on (tree_id, level, item_id): a retry after a failed + // cascade (step 2 of append_leaf) is a no-op, so duplicated children + // and double-counted tokens can't slip into the buffer. oldest_at + // stays on first-seen. + if buf.item_ids.iter().any(|existing| existing == item_id) { + log::debug!( + "[source_tree::bucket_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> { + 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)?; + // Forward-compat: index any entities the summariser emitted into + // `mem_tree_entity_index` so Phase 4 retrieval can resolve + // "summaries mentioning Alice" via the same inverted index as + // leaves. No-op under InertSummariser (entities is empty by + // design — see summariser/inert.rs doc); becomes active once 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 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..19d3ac24e --- /dev/null +++ b/src/openhuman/memory/tree/source_tree/flush.rs @@ -0,0 +1,190 @@ +//! 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..430c014cf --- /dev/null +++ b/src/openhuman/memory/tree/source_tree/registry.rs @@ -0,0 +1,172 @@ +//! 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, + }; + match store::insert_tree(config, &tree) { + Ok(()) => { + log::info!( + "[source_tree::registry] created source tree id={} scope={}", + tree.id, + scope + ); + Ok(tree) + } + Err(err) if is_unique_violation(&err) => { + // Race: another caller created a tree for the same scope + // between our initial lookup and this insert. UNIQUE(kind, + // scope) rejected our row; re-query and return the winner. + log::debug!( + "[source_tree::registry] UNIQUE race for scope={} — re-querying", + scope + ); + store::get_tree_by_scope(config, TreeKind::Source, scope)?.ok_or_else(|| { + anyhow::anyhow!( + "UNIQUE violation on insert but no row found on re-query for scope {scope}" + ) + }) + } + Err(err) => Err(err), + } +} + +/// Return true if `err` represents a SQLite UNIQUE constraint violation. +/// Matches both the anyhow-wrapped rusqlite error text and the raw SQLite +/// error codes in case the wrapping chain is shorter. +fn is_unique_violation(err: &anyhow::Error) -> bool { + if let Some(rusqlite_err) = err.downcast_ref::() { + if let rusqlite::Error::SqliteFailure(sqlite_err, _) = rusqlite_err { + return sqlite_err.code == rusqlite::ErrorCode::ConstraintViolation; + } + } + // Fallback for chained/wrapped errors: scan the rendered message. + let msg = format!("{err:#}"); + msg.contains("UNIQUE constraint failed") +} + +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:")); + } + + #[test] + fn get_or_create_recovers_from_unique_race() { + // Simulate the race by pre-inserting a tree under the same scope + // with a different id. `get_or_create` must re-query and return + // the pre-existing row, not bubble the UNIQUE error. + let (_tmp, cfg) = test_config(); + let pre_existing = Tree { + id: "source:preexisting".into(), + kind: TreeKind::Source, + scope: "slack:#eng".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(); + + // First call finds it via get_tree_by_scope (happy path — no race + // triggered here). To hit the race branch we need a caller that + // skips the lookup and goes straight to insert with a fresh id. + // Simplest proxy: call get_or_create twice from this test thread; + // the first creates, the second's UNIQUE would fire if the + // lookup was ever elided. Instead we cover the race path directly + // via `is_unique_violation` on a synthesised insert failure below. + let got = get_or_create_source_tree(&cfg, "slack:#eng").unwrap(); + assert_eq!(got.id, "source:preexisting"); + + // Direct coverage: a second insert with a different id for the + // same scope must surface as UNIQUE and be detected. + let dup = Tree { + id: "source: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/source_tree/store.rs b/src/openhuman/memory/tree/source_tree/store.rs new file mode 100644 index 000000000..82b57e2b7 --- /dev/null +++ b/src/openhuman/memory/tree/source_tree/store.rs @@ -0,0 +1,594 @@ +//! 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..f6f897113 --- /dev/null +++ b/src/openhuman/memory/tree/source_tree/summariser/inert.rs @@ -0,0 +1,176 @@ +//! Deterministic fallback summariser (#709). +//! +//! `InertSummariser` concatenates each input's content, separated by a +//! blank line, and hard-truncates to `ctx.token_budget`. Entities and +//! topics are **intentionally empty**: per design, summary-level entity / +//! topic metadata is derived by the LLM summariser from the summary's own +//! synthesised content (not by mechanically unioning children's labels). +//! Until the networked summariser lands, inert-sealed summaries have no +//! entity index rows — an honest stub. The goal of this fallback is not +//! metadata 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::{ + Summariser, SummaryContext, SummaryInput, SummaryOutput, +}; +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); + + log::debug!( + "[source_tree::summariser::inert] sealed tree_id={} level={} inputs={} tokens={} \ + entities=0 topics=0 (honest stub — LLM summariser derives these)", + ctx.tree_id, + ctx.target_level, + inputs.len(), + token_count + ); + + Ok(SummaryOutput { + content, + token_count, + entities: Vec::new(), + topics: Vec::new(), + }) + } +} + +/// 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) +} + +#[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 honest_stub_emits_no_entities_or_topics() { + // Per design: summary-level entities/topics are LLM-derived from + // the summary's own synthesised content. The inert fallback does + // not propagate children's labels — it emits empty vecs. The + // Ollama summariser (future) will fill them via real NER on its + // own output. + 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!(out.entities.is_empty()); + assert!(out.topics.is_empty()); + } + + #[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) }