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