diff --git a/src/openhuman/memory/tree/ingest.rs b/src/openhuman/memory/tree/ingest.rs index 448a41566..9cf81987e 100644 --- a/src/openhuman/memory/tree/ingest.rs +++ b/src/openhuman/memory/tree/ingest.rs @@ -24,6 +24,7 @@ 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::topic_tree::route_leaf_to_topic_trees; use crate::openhuman::memory::tree::types::Chunk; /// Outcome of one ingest call — extended with per-chunk admission info. @@ -237,7 +238,7 @@ async fn append_leaves_to_tree( r.canonical_entities .iter() .map(|e| e.canonical_id.clone()) - .collect(), + .collect::>(), chunk.metadata.tags.clone(), ), None => (0.0, Vec::new(), chunk.metadata.tags.clone()), @@ -247,11 +248,22 @@ async fn append_leaves_to_tree( token_count: chunk.token_count, timestamp: chunk.metadata.timestamp, content: chunk.content.clone(), - entities, + entities: entities.clone(), topics, score: score_value, }; append_leaf(config, &tree, &leaf, &summariser).await?; + + // Phase 3c (#709): route the leaf to every matching topic tree + // and tick the curator for each entity. Non-fatal on error — + // the source-tree append has already succeeded above. + if let Err(e) = route_leaf_to_topic_trees(config, &leaf, &entities, &summariser).await { + log::warn!( + "[memory_tree::ingest] topic_tree routing failed chunk_id={} err={:#}", + chunk.id, + e + ); + } } Ok(()) } diff --git a/src/openhuman/memory/tree/mod.rs b/src/openhuman/memory/tree/mod.rs index ba5d8037d..025411c52 100644 --- a/src/openhuman/memory/tree/mod.rs +++ b/src/openhuman/memory/tree/mod.rs @@ -31,6 +31,7 @@ pub mod schemas; pub mod score; pub mod source_tree; pub mod store; +pub mod topic_tree; pub mod types; pub use schemas::{ diff --git a/src/openhuman/memory/tree/store.rs b/src/openhuman/memory/tree/store.rs index 53e7fd7d8..0cbd8e99f 100644 --- a/src/openhuman/memory/tree/store.rs +++ b/src/openhuman/memory/tree/store.rs @@ -153,6 +153,27 @@ CREATE TABLE IF NOT EXISTS mem_tree_buffers ( CREATE INDEX IF NOT EXISTS idx_mem_tree_buffers_oldest ON mem_tree_buffers(oldest_at_ms); + +-- Phase 3c (#709): per-entity hotness counters driving lazy topic-tree +-- materialisation. One row per canonical entity_id. Counters are bumped +-- on every ingest; `last_hotness` is recomputed every +-- `TOPIC_RECHECK_EVERY` ingests to decide whether to spawn / archive a +-- topic tree for the entity. TODO: 30-day windowing — for Phase 3c we +-- increment counts forever and rely on project-scale truthfulness. +CREATE TABLE IF NOT EXISTS mem_tree_entity_hotness ( + entity_id TEXT PRIMARY KEY, + mention_count_30d INTEGER NOT NULL DEFAULT 0, + distinct_sources INTEGER NOT NULL DEFAULT 0, + last_seen_ms INTEGER, + query_hits_30d INTEGER NOT NULL DEFAULT 0, + graph_centrality REAL, + ingests_since_check INTEGER NOT NULL DEFAULT 0, + last_hotness REAL, + last_updated_ms INTEGER NOT NULL +); + +CREATE INDEX IF NOT EXISTS idx_mem_tree_entity_hotness_score + ON mem_tree_entity_hotness(last_hotness); "; /// Upsert a batch of chunks atomically. diff --git a/src/openhuman/memory/tree/topic_tree/backfill.rs b/src/openhuman/memory/tree/topic_tree/backfill.rs new file mode 100644 index 000000000..f018b4e11 --- /dev/null +++ b/src/openhuman/memory/tree/topic_tree/backfill.rs @@ -0,0 +1,304 @@ +//! Topic-tree backfill — hydrate a freshly-materialised topic tree with +//! every historical leaf mentioning the entity (#709 Phase 3c). +//! +//! When the curator decides an entity has crossed the hotness threshold +//! for the first time, we create a fresh topic tree AND walk the +//! `mem_tree_entity_index` inverted index to append every prior leaf into +//! its L0 buffer. Reusing `bucket_seal::append_leaf` means the cascade +//! fires automatically — a well-established entity may seal several +//! levels as soon as the tree is spawned. +//! +//! Backfill is intentionally best-effort: missing chunks are skipped with +//! a warn log rather than failing the whole spawn, because Phase 3c is +//! additive — a partial topic tree is still useful. + +use anyhow::{Context, Result}; + +use crate::openhuman::config::Config; +use crate::openhuman::memory::tree::score::store::lookup_entity; +use crate::openhuman::memory::tree::source_tree::bucket_seal::{append_leaf, LeafRef}; +use crate::openhuman::memory::tree::source_tree::summariser::Summariser; +use crate::openhuman::memory::tree::source_tree::types::Tree; +use crate::openhuman::memory::tree::store::get_chunk; + +/// Max leaves to pull from the entity index during backfill. A hard cap +/// keeps initial spawn latency bounded even for very active entities. +const BACKFILL_LIMIT: usize = 500; + +/// Walk the entity index for `entity_id` and append every discovered leaf +/// to `tree`. Returns the number of leaves appended (NOT the number of +/// summaries sealed). Idempotent: `append_leaf` itself is a no-op when a +/// leaf is already in the buffer, so re-running backfill is safe. +pub async fn backfill_topic_tree( + config: &Config, + tree: &Tree, + entity_id: &str, + summariser: &dyn Summariser, +) -> Result { + log::info!( + "[topic_tree::backfill] start entity_id={} tree_id={}", + entity_id, + tree.id + ); + + let hits = lookup_entity(config, entity_id, Some(BACKFILL_LIMIT)) + .with_context(|| format!("failed to lookup entity {entity_id}"))?; + + if hits.is_empty() { + log::debug!( + "[topic_tree::backfill] no entity-index hits for entity_id={} — empty backfill", + entity_id + ); + return Ok(0); + } + + // Sort by timestamp ASC so the buffer's `oldest_at` and the sealed + // summary's `time_range_start` reflect the true historical order, not + // the DESC ordering `lookup_entity` returns. + let mut hits = hits; + hits.sort_by_key(|h| h.timestamp_ms); + + let mut appended = 0usize; + for hit in hits { + // Skip summary-node hits — Phase 3c backfill only routes raw leaves + // into the topic tree. Including summary nodes would fold + // summaries-of-summaries across unrelated sources, which defeats + // the point. + if hit.node_kind != "leaf" { + log::debug!( + "[topic_tree::backfill] skipping non-leaf hit node_id={} kind={}", + hit.node_id, + hit.node_kind + ); + continue; + } + + let chunk = match get_chunk(config, &hit.node_id)? { + Some(c) => c, + None => { + log::warn!( + "[topic_tree::backfill] missing chunk {} for entity {} — skipping", + hit.node_id, + entity_id + ); + continue; + } + }; + + let leaf = LeafRef { + chunk_id: chunk.id.clone(), + token_count: chunk.token_count, + timestamp: chunk.metadata.timestamp, + content: chunk.content.clone(), + entities: vec![entity_id.to_string()], + topics: chunk.metadata.tags.clone(), + score: hit.score, + }; + + append_leaf(config, tree, &leaf, summariser) + .await + .with_context(|| { + format!( + "backfill append_leaf failed tree_id={} chunk_id={}", + tree.id, chunk.id + ) + })?; + appended += 1; + } + + log::info!( + "[topic_tree::backfill] done entity_id={} tree_id={} appended={}", + entity_id, + tree.id, + appended + ); + + Ok(appended) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::openhuman::memory::tree::score::extract::EntityKind; + use crate::openhuman::memory::tree::score::resolver::CanonicalEntity; + use crate::openhuman::memory::tree::score::store::index_entity; + use crate::openhuman::memory::tree::source_tree::store as src_store; + use crate::openhuman::memory::tree::source_tree::summariser::inert::InertSummariser; + use crate::openhuman::memory::tree::store::upsert_chunks; + use crate::openhuman::memory::tree::topic_tree::registry::get_or_create_topic_tree; + use crate::openhuman::memory::tree::types::{chunk_id, Chunk, Metadata, SourceKind, SourceRef}; + use chrono::{TimeZone, Utc}; + 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_chunk(source_id: &str, seq: u32, ts_ms: i64, tokens: u32) -> Chunk { + let ts = Utc.timestamp_millis_opt(ts_ms).unwrap(); + Chunk { + id: chunk_id(SourceKind::Chat, source_id, seq), + content: format!("substantive chunk mentioning alice {source_id}#{seq}"), + metadata: Metadata { + source_kind: SourceKind::Chat, + source_id: source_id.to_string(), + owner: "alice".into(), + timestamp: ts, + time_range: (ts, ts), + tags: vec!["eng".into()], + source_ref: Some(SourceRef::new(format!("{source_id}://{seq}"))), + }, + token_count: tokens, + seq_in_source: seq, + created_at: ts, + } + } + + fn sample_entity(canonical: &str, surface: &str) -> CanonicalEntity { + CanonicalEntity { + canonical_id: canonical.to_string(), + kind: EntityKind::Email, + surface: surface.to_string(), + span_start: 0, + span_end: surface.len() as u32, + score: 1.0, + } + } + + #[tokio::test] + async fn backfill_appends_all_entity_leaves() { + let (_tmp, cfg) = test_config(); + // Persist 3 chunks across 2 sources. + let c1 = mk_chunk("slack:#eng", 0, 1_700_000_000_000, 100); + let c2 = mk_chunk("gmail:alice", 0, 1_700_000_010_000, 100); + let c3 = mk_chunk("slack:#eng", 1, 1_700_000_020_000, 100); + upsert_chunks(&cfg, &[c1.clone(), c2.clone(), c3.clone()]).unwrap(); + + let e = sample_entity("email:alice@example.com", "alice@example.com"); + index_entity( + &cfg, + &e, + &c1.id, + "leaf", + 1_700_000_000_000, + Some("source:slack"), + ) + .unwrap(); + index_entity( + &cfg, + &e, + &c2.id, + "leaf", + 1_700_000_010_000, + Some("source:gmail"), + ) + .unwrap(); + index_entity( + &cfg, + &e, + &c3.id, + "leaf", + 1_700_000_020_000, + Some("source:slack"), + ) + .unwrap(); + + let tree = get_or_create_topic_tree(&cfg, "email:alice@example.com").unwrap(); + let summariser = InertSummariser::new(); + let n = backfill_topic_tree(&cfg, &tree, "email:alice@example.com", &summariser) + .await + .unwrap(); + assert_eq!(n, 3); + + // L0 buffer should hold all three leaves (combined tokens well + // under the 10k seal budget). + let buf = src_store::get_buffer(&cfg, &tree.id, 0).unwrap(); + assert_eq!(buf.item_ids.len(), 3); + assert_eq!(buf.token_sum, 300); + // Oldest item is c1. + assert_eq!(buf.oldest_at.unwrap().timestamp_millis(), 1_700_000_000_000); + } + + #[tokio::test] + async fn backfill_skips_missing_chunks_without_failing() { + let (_tmp, cfg) = test_config(); + let e = sample_entity("email:alice@example.com", "alice@example.com"); + // Index a chunk that was never persisted. + index_entity(&cfg, &e, "chunk:missing", "leaf", 1_700_000_000_000, None).unwrap(); + // And one that was. + let c = mk_chunk("slack:#eng", 0, 1_700_000_010_000, 100); + upsert_chunks(&cfg, &[c.clone()]).unwrap(); + index_entity( + &cfg, + &e, + &c.id, + "leaf", + 1_700_000_010_000, + Some("source:slack"), + ) + .unwrap(); + + let tree = get_or_create_topic_tree(&cfg, "email:alice@example.com").unwrap(); + let summariser = InertSummariser::new(); + let n = backfill_topic_tree(&cfg, &tree, "email:alice@example.com", &summariser) + .await + .unwrap(); + assert_eq!(n, 1, "only the existing chunk should be appended"); + let buf = src_store::get_buffer(&cfg, &tree.id, 0).unwrap(); + assert_eq!(buf.item_ids.len(), 1); + } + + #[tokio::test] + async fn backfill_is_idempotent() { + let (_tmp, cfg) = test_config(); + let c = mk_chunk("slack:#eng", 0, 1_700_000_000_000, 50); + upsert_chunks(&cfg, &[c.clone()]).unwrap(); + let e = sample_entity("email:alice@example.com", "alice@example.com"); + index_entity( + &cfg, + &e, + &c.id, + "leaf", + 1_700_000_000_000, + Some("source:slack"), + ) + .unwrap(); + + let tree = get_or_create_topic_tree(&cfg, "email:alice@example.com").unwrap(); + let summariser = InertSummariser::new(); + backfill_topic_tree(&cfg, &tree, "email:alice@example.com", &summariser) + .await + .unwrap(); + backfill_topic_tree(&cfg, &tree, "email:alice@example.com", &summariser) + .await + .unwrap(); + // append_leaf is idempotent so the buffer still has exactly one row. + let buf = src_store::get_buffer(&cfg, &tree.id, 0).unwrap(); + assert_eq!(buf.item_ids.len(), 1); + } + + #[tokio::test] + async fn backfill_skips_summary_nodes() { + let (_tmp, cfg) = test_config(); + let e = sample_entity("email:alice@example.com", "alice@example.com"); + // A summary-node hit in the entity index — should be skipped. + index_entity( + &cfg, + &e, + "summary:L1:abc", + "summary", + 1_700_000_000_000, + Some("source:slack"), + ) + .unwrap(); + let tree = get_or_create_topic_tree(&cfg, "email:alice@example.com").unwrap(); + let summariser = InertSummariser::new(); + let n = backfill_topic_tree(&cfg, &tree, "email:alice@example.com", &summariser) + .await + .unwrap(); + assert_eq!(n, 0); + } +} diff --git a/src/openhuman/memory/tree/topic_tree/curator.rs b/src/openhuman/memory/tree/topic_tree/curator.rs new file mode 100644 index 000000000..f5d69a2e3 --- /dev/null +++ b/src/openhuman/memory/tree/topic_tree/curator.rs @@ -0,0 +1,370 @@ +//! Topic-tree curator — the hotness gate (#709 Phase 3c). +//! +//! On every ingest that touches an entity we bump cheap counters +//! (`mention_count_30d`, `last_seen_ms`, `ingests_since_check`). Every +//! [`TOPIC_RECHECK_EVERY`] bumps we run the full hotness recompute: +//! +//! 1. Refresh `distinct_sources` from `mem_tree_entity_index`. +//! 2. Compute [`hotness`](super::hotness::hotness). +//! 3. If hotness ≥ [`TOPIC_CREATION_THRESHOLD`] and no topic tree exists +//! yet → create one and kick off [`backfill_topic_tree`]. +//! 4. Reset `ingests_since_check` to 0. +//! +//! The function is idempotent: if a topic tree already exists for the +//! entity it's a no-op at the creation step. Spawning is single-shot — +//! re-crossing the threshold after an archive would require explicit +//! unarchival (not Phase 3c). + +use anyhow::Result; +use chrono::Utc; + +use crate::openhuman::config::Config; +use crate::openhuman::memory::tree::source_tree::store as src_store; +use crate::openhuman::memory::tree::source_tree::summariser::Summariser; +use crate::openhuman::memory::tree::source_tree::types::{Tree, TreeKind}; +use crate::openhuman::memory::tree::topic_tree::backfill::backfill_topic_tree; +use crate::openhuman::memory::tree::topic_tree::hotness::hotness_at; +use crate::openhuman::memory::tree::topic_tree::registry::get_or_create_topic_tree; +use crate::openhuman::memory::tree::topic_tree::store::{ + distinct_sources_for, get_or_fresh, upsert, +}; +use crate::openhuman::memory::tree::topic_tree::types::{ + HotnessCounters, TOPIC_CREATION_THRESHOLD, TOPIC_RECHECK_EVERY, +}; + +/// Outcome of one curator invocation. Surfaced so the caller (typically +/// the routing layer) can log / emit metrics. +#[derive(Clone, Debug, PartialEq)] +pub enum SpawnOutcome { + /// Counters bumped; hotness not yet recomputed this round. + CountersBumped, + /// Full recompute ran; hotness below threshold, no tree spawned. + BelowThreshold { hotness: f32 }, + /// Tree already existed — just bumped counters and refreshed hotness. + TreeExists { hotness: f32, tree_id: String }, + /// Brand new topic tree materialised. + Spawned { + hotness: f32, + tree_id: String, + backfilled: usize, + }, +} + +/// Record an ingest touching `entity_id` and, when the recheck cadence +/// fires, consider spawning a topic tree. +/// +/// `summariser` is used only when a spawn + backfill happens; passing an +/// [`InertSummariser`](crate::openhuman::memory::tree::source_tree::summariser::inert::InertSummariser) +/// is fine for Phase 3c. +pub async fn maybe_spawn_topic_tree( + config: &Config, + entity_id: &str, + summariser: &dyn Summariser, +) -> Result { + let now_ms = Utc::now().timestamp_millis(); + + // 1. Read existing counters (fresh row if first sighting). + let mut counters = get_or_fresh(config, entity_id)?; + + // 2. Cheap per-ingest bumps. + counters.mention_count_30d = counters.mention_count_30d.saturating_add(1); + counters.last_seen_ms = Some(now_ms); + counters.ingests_since_check = counters.ingests_since_check.saturating_add(1); + counters.last_updated_ms = now_ms; + + // 3. Decide whether to run the full recompute. + if counters.ingests_since_check < TOPIC_RECHECK_EVERY { + upsert(config, &counters)?; + log::debug!( + "[topic_tree::curator] bumped counters entity={} mentions={} ingests_since_check={}", + entity_id, + counters.mention_count_30d, + counters.ingests_since_check + ); + return Ok(SpawnOutcome::CountersBumped); + } + + // 4. Full recompute. + run_full_recompute(config, entity_id, &mut counters, now_ms, summariser).await +} + +/// Admin path: force a recompute + spawn-if-hot regardless of the +/// [`TOPIC_RECHECK_EVERY`] cadence. Used by (future) RPCs that want to +/// prod the curator without waiting for the next bump cycle. +pub async fn force_recompute( + config: &Config, + entity_id: &str, + summariser: &dyn Summariser, +) -> Result { + let now_ms = Utc::now().timestamp_millis(); + let mut counters = get_or_fresh(config, entity_id)?; + counters.last_updated_ms = now_ms; + run_full_recompute(config, entity_id, &mut counters, now_ms, summariser).await +} + +async fn run_full_recompute( + config: &Config, + entity_id: &str, + counters: &mut HotnessCounters, + now_ms: i64, + summariser: &dyn Summariser, +) -> Result { + // Refresh distinct_sources from the entity index — the authoritative + // source of cross-tree coverage. + let distinct = distinct_sources_for(config, entity_id)?; + counters.distinct_sources = distinct; + + // Compute hotness against the refreshed stats. + let stats = counters.stats(); + let h = hotness_at(entity_id, &stats, now_ms); + + counters.last_hotness = Some(h); + counters.ingests_since_check = 0; + + let outcome = if h < TOPIC_CREATION_THRESHOLD { + log::debug!( + "[topic_tree::curator] below threshold entity={} hotness={:.3} threshold={}", + entity_id, + h, + TOPIC_CREATION_THRESHOLD + ); + SpawnOutcome::BelowThreshold { hotness: h } + } else if let Some(existing) = existing_topic_tree(config, entity_id)? { + log::debug!( + "[topic_tree::curator] tree already exists entity={} tree_id={} hotness={:.3}", + entity_id, + existing.id, + h + ); + SpawnOutcome::TreeExists { + hotness: h, + tree_id: existing.id, + } + } else { + // Crossed threshold for the first time — materialise. + log::info!( + "[topic_tree::curator] spawning topic tree entity={} hotness={:.3}", + entity_id, + h + ); + let tree = get_or_create_topic_tree(config, entity_id)?; + let backfilled = backfill_topic_tree(config, &tree, entity_id, summariser).await?; + SpawnOutcome::Spawned { + hotness: h, + tree_id: tree.id, + backfilled, + } + }; + + // Persist the refreshed counters regardless of outcome. + upsert(config, counters)?; + Ok(outcome) +} + +fn existing_topic_tree(config: &Config, entity_id: &str) -> Result> { + src_store::get_tree_by_scope(config, TreeKind::Topic, entity_id) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::openhuman::memory::tree::score::extract::EntityKind; + use crate::openhuman::memory::tree::score::resolver::CanonicalEntity; + use crate::openhuman::memory::tree::score::store::index_entity; + use crate::openhuman::memory::tree::source_tree::summariser::inert::InertSummariser; + use crate::openhuman::memory::tree::store::upsert_chunks; + use crate::openhuman::memory::tree::topic_tree::store::get; + use crate::openhuman::memory::tree::types::{chunk_id, Chunk, Metadata, SourceKind, SourceRef}; + use chrono::{TimeZone, Utc}; + 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 seed_leaf_for_entity(cfg: &Config, entity_id: &str, source_tree: &str, seq: u32) { + let ts_ms = 1_700_000_000_000 + (seq as i64) * 1_000; + let ts = Utc.timestamp_millis_opt(ts_ms).unwrap(); + let c = Chunk { + id: chunk_id(SourceKind::Chat, source_tree, seq), + content: format!("mentioning entity in {source_tree}#{seq}"), + metadata: Metadata { + source_kind: SourceKind::Chat, + source_id: source_tree.to_string(), + owner: "alice".into(), + timestamp: ts, + time_range: (ts, ts), + tags: vec![], + source_ref: Some(SourceRef::new(format!("{source_tree}://{seq}"))), + }, + token_count: 50, + seq_in_source: seq, + created_at: ts, + }; + upsert_chunks(cfg, &[c.clone()]).unwrap(); + let e = CanonicalEntity { + canonical_id: entity_id.to_string(), + kind: EntityKind::Email, + surface: entity_id.to_string(), + span_start: 0, + span_end: entity_id.len() as u32, + score: 1.0, + }; + index_entity(cfg, &e, &c.id, "leaf", ts_ms, Some(source_tree)).unwrap(); + } + + #[tokio::test] + async fn first_ingest_just_bumps_counters() { + let (_tmp, cfg) = test_config(); + let summariser = InertSummariser::new(); + let out = maybe_spawn_topic_tree(&cfg, "email:alice@example.com", &summariser) + .await + .unwrap(); + assert_eq!(out, SpawnOutcome::CountersBumped); + let c = get(&cfg, "email:alice@example.com").unwrap().unwrap(); + assert_eq!(c.mention_count_30d, 1); + assert_eq!(c.ingests_since_check, 1); + assert!(c.last_hotness.is_none(), "no recompute yet"); + } + + #[tokio::test] + async fn no_spawn_below_threshold_on_recompute() { + let (_tmp, cfg) = test_config(); + let summariser = InertSummariser::new(); + // Force a recompute on the very first call — but with no index data + // the hotness comes out well below threshold. + let out = force_recompute(&cfg, "email:alice@example.com", &summariser) + .await + .unwrap(); + match out { + SpawnOutcome::BelowThreshold { hotness } => { + assert!(hotness < TOPIC_CREATION_THRESHOLD); + } + other => panic!("expected BelowThreshold, got {other:?}"), + } + // No topic tree created. + let t = existing_topic_tree(&cfg, "email:alice@example.com").unwrap(); + assert!(t.is_none()); + } + + #[tokio::test] + async fn spawn_fires_exactly_once_when_threshold_crossed() { + let (_tmp, cfg) = test_config(); + let summariser = InertSummariser::new(); + // Seed substantial activity across several sources so hotness is + // well above threshold. + let mut counters = HotnessCounters::fresh("email:alice@example.com", 0); + counters.mention_count_30d = 500; + counters.distinct_sources = 4; + counters.last_seen_ms = Some(Utc::now().timestamp_millis()); + counters.query_hits_30d = 5; + upsert(&cfg, &counters).unwrap(); + // Seed leaves in the entity index so backfill has something to do. + for i in 0..3 { + seed_leaf_for_entity(&cfg, "email:alice@example.com", "slack:#eng", i); + } + for i in 0..2 { + seed_leaf_for_entity(&cfg, "email:alice@example.com", "gmail:alice", i); + } + + let out = force_recompute(&cfg, "email:alice@example.com", &summariser) + .await + .unwrap(); + match out { + SpawnOutcome::Spawned { + hotness, + tree_id, + backfilled, + } => { + assert!(hotness >= TOPIC_CREATION_THRESHOLD); + assert!(tree_id.starts_with("topic:")); + assert_eq!(backfilled, 5); + } + other => panic!("expected Spawned, got {other:?}"), + } + + // Re-running should report TreeExists, NOT a second spawn. + let out2 = force_recompute(&cfg, "email:alice@example.com", &summariser) + .await + .unwrap(); + match out2 { + SpawnOutcome::TreeExists { tree_id, .. } => { + assert!(tree_id.starts_with("topic:")); + } + other => panic!("expected TreeExists on retry, got {other:?}"), + } + } + + #[tokio::test] + async fn recompute_refreshes_distinct_sources_from_entity_index() { + let (_tmp, cfg) = test_config(); + let summariser = InertSummariser::new(); + // Counter says 0 distinct sources but the index has 3. + let mut counters = HotnessCounters::fresh("email:alice@example.com", 0); + counters.mention_count_30d = 1; + counters.distinct_sources = 0; + upsert(&cfg, &counters).unwrap(); + seed_leaf_for_entity(&cfg, "email:alice@example.com", "slack:#eng", 0); + seed_leaf_for_entity(&cfg, "email:alice@example.com", "gmail:alice", 0); + seed_leaf_for_entity(&cfg, "email:alice@example.com", "notion:abc", 0); + + force_recompute(&cfg, "email:alice@example.com", &summariser) + .await + .unwrap(); + let c = get(&cfg, "email:alice@example.com").unwrap().unwrap(); + assert_eq!(c.distinct_sources, 3); + // ingests_since_check should also reset. + assert_eq!(c.ingests_since_check, 0); + } + + #[tokio::test] + async fn cadence_only_recomputes_every_n_ingests() { + let (_tmp, cfg) = test_config(); + let summariser = InertSummariser::new(); + // Pre-seed entity index with enough cross-source signal that the + // recompute (which refreshes `distinct_sources` from the index) will + // still produce a hotness above threshold. + for i in 0..4 { + seed_leaf_for_entity( + &cfg, + "email:alice@example.com", + &format!("slack:#eng-{i}"), + i, + ); + } + + let mut counters = HotnessCounters::fresh("email:alice@example.com", 0); + counters.mention_count_30d = 500; + counters.distinct_sources = 4; + counters.last_seen_ms = Some(Utc::now().timestamp_millis()); + // Boost query_hits so hotness stays comfortably above threshold + // after the distinct_sources refresh. + counters.query_hits_30d = 5; + // ingests_since_check just below the cadence: next call should + // NOT yet recompute. + counters.ingests_since_check = TOPIC_RECHECK_EVERY - 2; + upsert(&cfg, &counters).unwrap(); + + let out = maybe_spawn_topic_tree(&cfg, "email:alice@example.com", &summariser) + .await + .unwrap(); + assert_eq!(out, SpawnOutcome::CountersBumped); + // No tree yet — cadence not crossed. + assert!(existing_topic_tree(&cfg, "email:alice@example.com") + .unwrap() + .is_none()); + + // One more bump — now ingests_since_check == TOPIC_RECHECK_EVERY + // and the recompute fires. + let out2 = maybe_spawn_topic_tree(&cfg, "email:alice@example.com", &summariser) + .await + .unwrap(); + match out2 { + SpawnOutcome::Spawned { .. } | SpawnOutcome::TreeExists { .. } => {} + other => panic!("expected Spawn/TreeExists after cadence, got {other:?}"), + } + } +} diff --git a/src/openhuman/memory/tree/topic_tree/hotness.rs b/src/openhuman/memory/tree/topic_tree/hotness.rs new file mode 100644 index 000000000..e757dcccc --- /dev/null +++ b/src/openhuman/memory/tree/topic_tree/hotness.rs @@ -0,0 +1,203 @@ +//! Pure hotness math for Phase 3c (#709). +//! +//! The formula intentionally folds a handful of pre-existing signals into +//! one arithmetic score. No LLM, no learned weights — the goal is +//! deterministic, greppable, testable behaviour: +//! +//! ```text +//! hotness = ln(mentions + 1) // dampened high-volume bias +//! + 0.5 * distinct_sources // cross-source is valuable +//! + recency_decay(last_seen) // prefer active entities +//! + graph_centrality // Phase 4+ (None → 0.0) +//! + 2.0 * query_hits // retrieval feedback (Phase 4+) +//! ``` +//! +//! Recency decay is a piecewise linear taper: +//! - age ≤ 1 day → 1.0 +//! - age 1…7 days → 1.0 → 0.5 +//! - age 7…30 days → 0.5 → 0.0 +//! - age > 30 days → 0.0 +//! +//! The unit tests lock in the coarse behaviour (zero-mention, spike, +//! old-but-widely-cited) so tuning the constants later stays honest. + +use chrono::Utc; + +use crate::openhuman::memory::tree::topic_tree::types::EntityIndexStats; + +/// Pure hotness function — no I/O, no clocks unless the caller passes one. +/// +/// `entity_id` is taken for diagnostic logging only and has no effect on +/// the numeric result. +pub fn hotness(entity_id: &str, idx: &EntityIndexStats) -> f32 { + let now_ms = Utc::now().timestamp_millis(); + hotness_at(entity_id, idx, now_ms) +} + +/// Deterministic variant — computes hotness as if the current wall clock +/// were `now_ms`. Useful in tests so the recency term doesn't drift. +pub fn hotness_at(entity_id: &str, idx: &EntityIndexStats, now_ms: i64) -> f32 { + let mention_weight = ((idx.mention_count_30d as f32) + 1.0).ln(); + let source_weight = (idx.distinct_sources as f32) * 0.5; + let recency_weight = recency_decay(idx.last_seen_ms, now_ms); + let centrality = idx.graph_centrality.unwrap_or(0.0); + let query_weight = (idx.query_hits_30d as f32) * 2.0; + + let total = mention_weight + source_weight + recency_weight + centrality + query_weight; + log::debug!( + "[topic_tree::hotness] id={} mentions={} sources={} recency={:.3} centrality={:.3} \ + queries={} total={:.3}", + entity_id, + idx.mention_count_30d, + idx.distinct_sources, + recency_weight, + centrality, + idx.query_hits_30d, + total + ); + total +} + +/// Recency decay helper. Operates on absolute epoch-millis so tests can +/// pin the clock. Returns 0.0 when `last_seen_ms` is `None`. +pub fn recency_decay(last_seen_ms: Option, now_ms: i64) -> f32 { + let Some(last_seen) = last_seen_ms else { + return 0.0; + }; + let age_ms = (now_ms - last_seen).max(0); + const DAY_MS: i64 = 24 * 60 * 60 * 1_000; + let age_days = (age_ms as f32) / (DAY_MS as f32); + + if age_days <= 1.0 { + 1.0 + } else if age_days <= 7.0 { + // 1.0 at day 1, 0.5 at day 7 + let frac = (age_days - 1.0) / 6.0; + 1.0 - 0.5 * frac + } else if age_days <= 30.0 { + // 0.5 at day 7, 0.0 at day 30 + let frac = (age_days - 7.0) / 23.0; + 0.5 - 0.5 * frac + } else { + 0.0 + } +} + +#[cfg(test)] +mod tests { + use super::*; + + const DAY_MS: i64 = 24 * 60 * 60 * 1_000; + + fn stats(mentions: u32, sources: u32, last_seen: Option) -> EntityIndexStats { + EntityIndexStats { + mention_count_30d: mentions, + distinct_sources: sources, + last_seen_ms: last_seen, + query_hits_30d: 0, + graph_centrality: None, + } + } + + #[test] + fn zero_signal_entity_is_zero() { + let now_ms = 1_700_000_000_000; + let s = stats(0, 0, None); + let h = hotness_at("e:none", &s, now_ms); + // ln(0+1) + 0 + 0 + 0 + 0 = 0 + assert!(h.abs() < 1e-6); + } + + #[test] + fn spike_of_mentions_pushes_over_creation_threshold() { + use crate::openhuman::memory::tree::topic_tree::types::TOPIC_CREATION_THRESHOLD; + let now_ms = 1_700_000_000_000; + // 100 mentions across 5 sources, 3 recent query hits, seen today. + let s = EntityIndexStats { + mention_count_30d: 100, + distinct_sources: 5, + last_seen_ms: Some(now_ms - DAY_MS / 2), + query_hits_30d: 3, + graph_centrality: None, + }; + let h = hotness_at("e:hot", &s, now_ms); + assert!( + h > TOPIC_CREATION_THRESHOLD, + "expected hot entity > {TOPIC_CREATION_THRESHOLD}, got {h}" + ); + } + + #[test] + fn old_but_widely_cited_still_has_some_heat() { + // 50 mentions, 8 sources, last seen 20 days ago, no queries. + let now_ms = 1_700_000_000_000; + let s = EntityIndexStats { + mention_count_30d: 50, + distinct_sources: 8, + last_seen_ms: Some(now_ms - 20 * DAY_MS), + query_hits_30d: 0, + graph_centrality: None, + }; + let h = hotness_at("e:old-wide", &s, now_ms); + // mention_weight = ln(51) ≈ 3.93, source_weight = 4.0, + // recency at day 20 ≈ 0.5 * (30-20)/23 ≈ 0.217 → total ≈ 8.1 + assert!(h > 5.0, "widely-cited entity should retain signal: {h}"); + } + + #[test] + fn ancient_single_mention_decays_toward_zero() { + let now_ms = 1_700_000_000_000; + let s = stats(1, 1, Some(now_ms - 60 * DAY_MS)); + let h = hotness_at("e:ancient", &s, now_ms); + // ln(2) + 0.5 + 0 = ~1.19 — well below creation threshold + assert!(h < 2.0, "ancient entity should decay: {h}"); + } + + #[test] + fn recency_decay_today_is_one() { + let now_ms = 1_700_000_000_000; + let r = recency_decay(Some(now_ms), now_ms); + assert!((r - 1.0).abs() < 1e-6); + } + + #[test] + fn recency_decay_week_old_is_half() { + let now_ms = 1_700_000_000_000; + let r = recency_decay(Some(now_ms - 7 * DAY_MS), now_ms); + assert!((r - 0.5).abs() < 1e-3, "expected 0.5 at 7d, got {r}"); + } + + #[test] + fn recency_decay_month_old_is_zero() { + let now_ms = 1_700_000_000_000; + let r = recency_decay(Some(now_ms - 30 * DAY_MS), now_ms); + assert!(r.abs() < 1e-3, "expected ~0 at 30d, got {r}"); + } + + #[test] + fn recency_decay_none_last_seen_is_zero() { + assert_eq!(recency_decay(None, 1_700_000_000_000), 0.0); + } + + #[test] + fn query_hits_boost_hotness_aggressively() { + let now_ms = 1_700_000_000_000; + let base = stats(5, 1, Some(now_ms)); + let boosted = EntityIndexStats { + query_hits_30d: 10, + ..base.clone() + }; + let h_base = hotness_at("e", &base, now_ms); + let h_boosted = hotness_at("e", &boosted, now_ms); + // 10 query hits * 2.0 = +20 + assert!(h_boosted - h_base > 19.0); + } + + #[test] + fn future_last_seen_is_treated_as_now() { + // Clock drift could produce negative ages — we clamp at 0. + let now_ms = 1_700_000_000_000; + let r = recency_decay(Some(now_ms + DAY_MS), now_ms); + assert!((r - 1.0).abs() < 1e-6); + } +} diff --git a/src/openhuman/memory/tree/topic_tree/mod.rs b/src/openhuman/memory/tree/topic_tree/mod.rs new file mode 100644 index 000000000..5f216005b --- /dev/null +++ b/src/openhuman/memory/tree/topic_tree/mod.rs @@ -0,0 +1,46 @@ +//! Phase 3c — topic trees (lazy materialisation) (#709). +//! +//! A *topic tree* is a per-entity summary tree whose leaves are all +//! chunks mentioning that entity, regardless of the source they came +//! from. Topic trees are spawned lazily — only when an entity's hotness +//! crosses a threshold — and then receive new leaves via the ingest path +//! alongside the per-source tree. See `docs/MEMORY_ARCHITECTURE_LLD.md` +//! for the full design. +//! +//! Phase 3c surface: +//! - [`curator::maybe_spawn_topic_tree`] — per-ingest tick; bumps +//! counters and spawns a topic tree when hotness crosses +//! [`types::TOPIC_CREATION_THRESHOLD`]. +//! - [`routing::route_leaf_to_topic_trees`] — called by the ingest path +//! after the source-tree append; fans a kept leaf out to every +//! matching entity's topic tree. +//! - [`registry::get_or_create_topic_tree`] / +//! [`registry::archive_topic_tree`] — primitives for admin flows. +//! - [`backfill::backfill_topic_tree`] — walk the entity index and +//! hydrate a freshly spawned tree with historic leaves. +//! - [`hotness::hotness`] — pure arithmetic over pre-existing signals; +//! easy to unit-test. +//! +//! Tree mechanics (buffer, seal, cascade) are **not reimplemented** here +//! — `append_leaf` from [`super::source_tree::bucket_seal`] takes a +//! `&Tree` so it works for any `TreeKind`. The Phase 3c code only adds +//! the hotness layer and the per-entity fan-out. + +pub mod backfill; +pub mod curator; +pub mod hotness; +pub mod registry; +pub mod routing; +pub mod store; +pub mod types; + +pub use curator::{maybe_spawn_topic_tree, SpawnOutcome}; +pub use hotness::{hotness, recency_decay}; +pub use registry::{ + archive_topic_tree, force_create_topic_tree, get_or_create_topic_tree, list_topic_trees, +}; +pub use routing::route_leaf_to_topic_trees; +pub use types::{ + EntityIndexStats, HotnessCounters, TOPIC_ARCHIVE_THRESHOLD, TOPIC_CREATION_THRESHOLD, + TOPIC_RECHECK_EVERY, +}; diff --git a/src/openhuman/memory/tree/topic_tree/registry.rs b/src/openhuman/memory/tree/topic_tree/registry.rs new file mode 100644 index 000000000..9002b8cdd --- /dev/null +++ b/src/openhuman/memory/tree/topic_tree/registry.rs @@ -0,0 +1,289 @@ +//! Topic tree registry — get-or-create / archive (#709 Phase 3c). +//! +//! Topic trees share the same `mem_tree_trees` schema as source trees; the +//! only difference is `kind = 'topic'` and `scope = `. +//! Callers should NOT reach into this module to create topic trees +//! eagerly — use the curator ([`super::curator::maybe_spawn_topic_tree`]) +//! so creation is gated on hotness. Admin flows (future RPC) that want to +//! bypass the gate can call [`force_create_topic_tree`] directly. + +use anyhow::{Context, 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 topic tree for `entity_id`, or create a new one. +/// +/// The `entity_id` is a canonical id from the entity resolver (e.g. +/// `"email:alice@example.com"` or `"hashtag:launch"`). Scope uses the +/// canonical id verbatim so re-lookups are stable. +pub fn get_or_create_topic_tree(config: &Config, entity_id: &str) -> Result { + if let Some(existing) = store::get_tree_by_scope(config, TreeKind::Topic, entity_id)? { + log::debug!( + "[topic_tree::registry] found tree id={} entity={}", + existing.id, + entity_id + ); + return Ok(existing); + } + create_new(config, entity_id) +} + +/// Public alias used by the admin "force materialise" path — semantically +/// identical to [`get_or_create_topic_tree`] but named to make intent at +/// the call site obvious. +pub fn force_create_topic_tree(config: &Config, entity_id: &str) -> Result { + get_or_create_topic_tree(config, entity_id) +} + +/// List all topic trees (both active and archived). Ordered by creation time +/// ascending for stable output. +pub fn list_topic_trees(config: &Config) -> Result> { + use rusqlite::params; + crate::openhuman::memory::tree::store::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![TreeKind::Topic.as_str()], row_to_tree_loose)? + .collect::>>() + .context("failed to list topic trees")?; + Ok(rows) + }) +} + +/// Flip a topic tree's status to `archived`. Existing rows remain queryable; +/// new leaves will NOT be routed to this tree until it's manually unarchived +/// (unarchive is not a Phase 3c primitive — Phase 3c just stops routing). +pub fn archive_topic_tree(config: &Config, tree_id: &str) -> Result<()> { + use rusqlite::params; + crate::openhuman::memory::tree::store::with_connection(config, |conn| { + let n = conn + .execute( + "UPDATE mem_tree_trees + SET status = ?1 + WHERE id = ?2 AND kind = ?3", + params![ + TreeStatus::Archived.as_str(), + tree_id, + TreeKind::Topic.as_str() + ], + ) + .with_context(|| format!("failed to archive topic tree {tree_id}"))?; + if n == 0 { + log::warn!( + "[topic_tree::registry] archive_topic_tree: no topic tree with id={tree_id}" + ); + } else { + log::info!("[topic_tree::registry] archived topic tree id={tree_id}"); + } + Ok(()) + }) +} + +fn create_new(config: &Config, entity_id: &str) -> Result { + let tree = Tree { + id: new_topic_tree_id(), + kind: TreeKind::Topic, + scope: entity_id.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!( + "[topic_tree::registry] created topic tree id={} entity={}", + tree.id, + entity_id + ); + Ok(tree) + } + Err(err) if is_unique_violation(&err) => { + log::debug!( + "[topic_tree::registry] UNIQUE race for entity={} — re-querying", + entity_id + ); + store::get_tree_by_scope(config, TreeKind::Topic, entity_id)?.ok_or_else(|| { + anyhow::anyhow!( + "UNIQUE violation on insert but no row found on re-query for entity {entity_id}" + ) + }) + } + Err(err) => Err(err), + } +} + +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; + } + } + let msg = format!("{err:#}"); + msg.contains("UNIQUE constraint failed") +} + +fn new_topic_tree_id() -> String { + format!("{}:{}", TreeKind::Topic.as_str(), Uuid::new_v4()) +} + +/// Row mapper — duplicated from `source_tree::store::row_to_tree` because +/// that one is private. Kept intentionally loose: topic-tree listing is +/// not a hot path so the string parsing cost is immaterial. +fn row_to_tree_loose(row: &rusqlite::Row<'_>) -> rusqlite::Result { + use chrono::TimeZone; + 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()) + })?; + let created_at = Utc + .timestamp_millis_opt(created_ms) + .single() + .ok_or_else(|| { + rusqlite::Error::FromSqlConversionFailure( + 6, + rusqlite::types::Type::Integer, + format!("invalid created_at_ms {created_ms}").into(), + ) + })?; + let last_sealed_at = last_sealed_ms + .map(|ms| { + Utc.timestamp_millis_opt(ms).single().ok_or_else(|| { + rusqlite::Error::FromSqlConversionFailure( + 7, + rusqlite::types::Type::Integer, + format!("invalid last_sealed_at_ms {ms}").into(), + ) + }) + }) + .transpose()?; + + Ok(Tree { + id, + kind, + scope, + root_id, + max_level: max_level.max(0) as u32, + status, + created_at, + last_sealed_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) + } + + #[test] + fn get_or_create_is_idempotent_on_entity_id() { + let (_tmp, cfg) = test_config(); + let first = get_or_create_topic_tree(&cfg, "email:alice@example.com").unwrap(); + let second = get_or_create_topic_tree(&cfg, "email:alice@example.com").unwrap(); + assert_eq!(first.id, second.id); + assert_eq!(first.kind, TreeKind::Topic); + assert_eq!(first.status, TreeStatus::Active); + assert_eq!(first.scope, "email:alice@example.com"); + } + + #[test] + fn different_entities_yield_different_trees() { + let (_tmp, cfg) = test_config(); + let a = get_or_create_topic_tree(&cfg, "email:alice@example.com").unwrap(); + let b = get_or_create_topic_tree(&cfg, "email:bob@example.com").unwrap(); + assert_ne!(a.id, b.id); + assert_ne!(a.scope, b.scope); + } + + #[test] + fn topic_tree_and_source_tree_share_scope_space_cleanly() { + // A source tree and a topic tree can have the same *logical* + // scope string (e.g. an entity id that looks like a source id) — + // the UNIQUE constraint is on (kind, scope), not scope alone. + let (_tmp, cfg) = test_config(); + let source = + crate::openhuman::memory::tree::source_tree::registry::get_or_create_source_tree( + &cfg, + "shared:slack:#eng", + ) + .unwrap(); + let topic = get_or_create_topic_tree(&cfg, "shared:slack:#eng").unwrap(); + assert_ne!(source.id, topic.id); + assert_eq!(source.kind, TreeKind::Source); + assert_eq!(topic.kind, TreeKind::Topic); + } + + #[test] + fn topic_tree_id_has_expected_prefix() { + let id = new_topic_tree_id(); + assert!(id.starts_with("topic:")); + } + + #[test] + fn archive_flips_status_and_keeps_rows_readable() { + let (_tmp, cfg) = test_config(); + let t = get_or_create_topic_tree(&cfg, "email:alice@example.com").unwrap(); + archive_topic_tree(&cfg, &t.id).unwrap(); + let refetched = store::get_tree(&cfg, &t.id).unwrap().unwrap(); + assert_eq!(refetched.status, TreeStatus::Archived); + // get_or_create should still return the same (archived) row rather + // than creating a new one — archiving is NOT deletion. + let again = get_or_create_topic_tree(&cfg, "email:alice@example.com").unwrap(); + assert_eq!(again.id, t.id); + assert_eq!(again.status, TreeStatus::Archived); + } + + #[test] + fn archive_is_noop_on_nonexistent() { + let (_tmp, cfg) = test_config(); + // Shouldn't error — just log a warning. + archive_topic_tree(&cfg, "topic:does-not-exist").unwrap(); + } + + #[test] + fn list_topic_trees_returns_only_topics() { + let (_tmp, cfg) = test_config(); + // Mix of source + topic trees. + crate::openhuman::memory::tree::source_tree::registry::get_or_create_source_tree( + &cfg, + "slack:#eng", + ) + .unwrap(); + get_or_create_topic_tree(&cfg, "email:alice@example.com").unwrap(); + get_or_create_topic_tree(&cfg, "email:bob@example.com").unwrap(); + + let topics = list_topic_trees(&cfg).unwrap(); + assert_eq!(topics.len(), 2); + for t in &topics { + assert_eq!(t.kind, TreeKind::Topic); + } + } +} diff --git a/src/openhuman/memory/tree/topic_tree/routing.rs b/src/openhuman/memory/tree/topic_tree/routing.rs new file mode 100644 index 000000000..f57e99dea --- /dev/null +++ b/src/openhuman/memory/tree/topic_tree/routing.rs @@ -0,0 +1,348 @@ +//! Per-leaf routing into topic trees (#709 Phase 3c). +//! +//! This is the hook point the ingest path calls after it has finished +//! appending a leaf to its source tree. For each canonical entity on the +//! chunk we: +//! +//! 1. Append the leaf to that entity's topic tree *if* one already exists +//! (active status only — archived topic trees don't receive new +//! leaves). +//! 2. Notify the curator that this entity was just mentioned, which may +//! cross the hotness threshold and spawn a new topic tree. +//! +//! Steps 1 and 2 are independent — if an entity's topic tree already +//! exists, step 2 just bumps counters; if it doesn't, step 1 is skipped +//! and step 2 may materialise it on this ingest. +//! +//! Failures are logged at warn level but never bubble up: Phase 3c is +//! additive and must not poison the ingest path. The source-tree append +//! has already succeeded by the time we get here. + +use anyhow::Result; + +use crate::openhuman::config::Config; +use crate::openhuman::memory::tree::source_tree::bucket_seal::{append_leaf, LeafRef}; +use crate::openhuman::memory::tree::source_tree::store as src_store; +use crate::openhuman::memory::tree::source_tree::summariser::Summariser; +use crate::openhuman::memory::tree::source_tree::types::{TreeKind, TreeStatus}; +use crate::openhuman::memory::tree::topic_tree::curator::maybe_spawn_topic_tree; + +/// Route `leaf` into every active topic tree matching one of +/// `canonical_entities`. Also ticks the curator for each entity so the +/// next cadence-aligned ingest may spawn a new tree. +/// +/// Returns `Ok(())` even if individual entities fail — per-entity errors +/// are logged. A hard DB failure early in the process is surfaced so the +/// caller can decide how loud to be in logs. +pub async fn route_leaf_to_topic_trees( + config: &Config, + leaf: &LeafRef, + canonical_entities: &[String], + summariser: &dyn Summariser, +) -> Result<()> { + if canonical_entities.is_empty() { + return Ok(()); + } + + log::debug!( + "[topic_tree::routing] leaf={} entities={}", + leaf.chunk_id, + canonical_entities.len() + ); + + for entity_id in canonical_entities { + if let Err(e) = route_one_entity(config, leaf, entity_id, summariser).await { + log::warn!( + "[topic_tree::routing] failed routing leaf={} entity={} err={:#}", + leaf.chunk_id, + entity_id, + e + ); + } + } + Ok(()) +} + +async fn route_one_entity( + config: &Config, + leaf: &LeafRef, + entity_id: &str, + summariser: &dyn Summariser, +) -> Result<()> { + // Step 1: if a topic tree already exists and is active, append the leaf. + // We intentionally do this BEFORE asking the curator to spawn — a + // same-call spawn would also include this leaf via backfill + // (`lookup_entity` was just updated by the ingest's score persist) but + // keeping the existing-tree fast path separate keeps the common case + // (hot entity already has a tree) clean. + if let Some(tree) = src_store::get_tree_by_scope(config, TreeKind::Topic, entity_id)? { + if tree.status == TreeStatus::Active { + log::debug!( + "[topic_tree::routing] appending leaf={} → topic_tree={}", + leaf.chunk_id, + tree.id + ); + // Rebuild the leaf with this entity-id stamped on so the seal + // path sees the topic membership. The source-tree append used + // the full entity list; here we scope to just this entity so + // the curated summariser (future) can prompt accordingly. + let topic_leaf = LeafRef { + entities: vec![entity_id.to_string()], + ..leaf.clone() + }; + append_leaf(config, &tree, &topic_leaf, summariser).await?; + } else { + log::debug!( + "[topic_tree::routing] skip archived topic tree id={} entity={}", + tree.id, + entity_id + ); + } + } + + // Step 2: curator tick — may spawn a new tree on cadence. + maybe_spawn_topic_tree(config, entity_id, summariser).await?; + + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::openhuman::memory::tree::score::extract::EntityKind; + use crate::openhuman::memory::tree::score::resolver::CanonicalEntity; + use crate::openhuman::memory::tree::score::store::index_entity; + use crate::openhuman::memory::tree::source_tree::summariser::inert::InertSummariser; + use crate::openhuman::memory::tree::store::upsert_chunks; + use crate::openhuman::memory::tree::topic_tree::registry::{ + archive_topic_tree, get_or_create_topic_tree, + }; + use crate::openhuman::memory::tree::topic_tree::store::get as get_hotness; + use crate::openhuman::memory::tree::types::{chunk_id, Chunk, Metadata, SourceKind, SourceRef}; + use chrono::{TimeZone, Utc}; + 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(chunk_id_s: &str, tokens: u32, ts_ms: i64) -> LeafRef { + LeafRef { + chunk_id: chunk_id_s.to_string(), + token_count: tokens, + timestamp: Utc.timestamp_millis_opt(ts_ms).unwrap(), + content: format!("content for {chunk_id_s}"), + entities: vec!["email:alice@example.com".into()], + topics: vec![], + score: 0.5, + } + } + + fn persist_chunk(cfg: &Config, source_id: &str, seq: u32, ts_ms: i64, tokens: u32) -> String { + let ts = Utc.timestamp_millis_opt(ts_ms).unwrap(); + let c = Chunk { + id: chunk_id(SourceKind::Chat, source_id, seq), + content: format!("chunk content {source_id} {seq}"), + metadata: Metadata { + source_kind: SourceKind::Chat, + source_id: source_id.to_string(), + owner: "alice".into(), + timestamp: ts, + time_range: (ts, ts), + tags: vec![], + source_ref: Some(SourceRef::new(format!("{source_id}://{seq}"))), + }, + token_count: tokens, + seq_in_source: seq, + created_at: ts, + }; + let id = c.id.clone(); + upsert_chunks(cfg, &[c]).unwrap(); + id + } + + #[tokio::test] + async fn empty_entities_is_noop() { + let (_tmp, cfg) = test_config(); + let summariser = InertSummariser::new(); + let leaf = mk_leaf("c1", 10, 1_700_000_000_000); + route_leaf_to_topic_trees(&cfg, &leaf, &[], &summariser) + .await + .unwrap(); + // No hotness rows were created. + assert_eq!( + crate::openhuman::memory::tree::topic_tree::store::count(&cfg).unwrap(), + 0 + ); + } + + #[tokio::test] + async fn appends_to_existing_topic_tree() { + let (_tmp, cfg) = test_config(); + let summariser = InertSummariser::new(); + // Pre-create the topic tree so the hot-path append fires. + let tree = get_or_create_topic_tree(&cfg, "email:alice@example.com").unwrap(); + // Persist the backing chunk so hydrate can read it on seal. + let chunk_id_s = persist_chunk(&cfg, "slack:#eng", 0, 1_700_000_000_000, 100); + let leaf = mk_leaf(&chunk_id_s, 100, 1_700_000_000_000); + + route_leaf_to_topic_trees( + &cfg, + &leaf, + &["email:alice@example.com".to_string()], + &summariser, + ) + .await + .unwrap(); + + let buf = src_store::get_buffer(&cfg, &tree.id, 0).unwrap(); + assert_eq!(buf.item_ids.len(), 1); + assert_eq!(buf.item_ids[0], chunk_id_s); + // Counter should also be bumped. + let c = get_hotness(&cfg, "email:alice@example.com") + .unwrap() + .unwrap(); + assert_eq!(c.mention_count_30d, 1); + } + + #[tokio::test] + async fn archived_topic_tree_does_not_receive_new_leaves() { + let (_tmp, cfg) = test_config(); + let summariser = InertSummariser::new(); + let tree = get_or_create_topic_tree(&cfg, "email:alice@example.com").unwrap(); + archive_topic_tree(&cfg, &tree.id).unwrap(); + + let chunk_id_s = persist_chunk(&cfg, "slack:#eng", 0, 1_700_000_000_000, 100); + let leaf = mk_leaf(&chunk_id_s, 100, 1_700_000_000_000); + route_leaf_to_topic_trees( + &cfg, + &leaf, + &["email:alice@example.com".to_string()], + &summariser, + ) + .await + .unwrap(); + + let buf = src_store::get_buffer(&cfg, &tree.id, 0).unwrap(); + assert!( + buf.is_empty(), + "archived topic tree should not receive new leaves" + ); + // Counter should still be bumped — archiving doesn't freeze hotness. + let c = get_hotness(&cfg, "email:alice@example.com") + .unwrap() + .unwrap(); + assert_eq!(c.mention_count_30d, 1); + } + + #[tokio::test] + async fn one_leaf_multiple_entities_fans_out() { + let (_tmp, cfg) = test_config(); + let summariser = InertSummariser::new(); + let t1 = get_or_create_topic_tree(&cfg, "email:alice@example.com").unwrap(); + let t2 = get_or_create_topic_tree(&cfg, "hashtag:launch").unwrap(); + + let chunk_id_s = persist_chunk(&cfg, "slack:#eng", 0, 1_700_000_000_000, 100); + let leaf = mk_leaf(&chunk_id_s, 100, 1_700_000_000_000); + route_leaf_to_topic_trees( + &cfg, + &leaf, + &[ + "email:alice@example.com".to_string(), + "hashtag:launch".to_string(), + ], + &summariser, + ) + .await + .unwrap(); + + // Both topic trees' L0 buffers hold the leaf. + let b1 = src_store::get_buffer(&cfg, &t1.id, 0).unwrap(); + let b2 = src_store::get_buffer(&cfg, &t2.id, 0).unwrap(); + assert_eq!(b1.item_ids.len(), 1); + assert_eq!(b2.item_ids.len(), 1); + } + + #[tokio::test] + async fn integration_two_sources_mentioning_alice_materialise_topic_tree() { + // Phase 3c acceptance scenario: ingest across 2 sources mentioning + // Alice → hotness crosses threshold → topic tree materialised → + // new Alice-mentioning leaf routes into both the source tree AND + // the topic tree. + let (_tmp, cfg) = test_config(); + let summariser = InertSummariser::new(); + let entity_id = "email:alice@example.com"; + + // Pre-seed counters / index so the next call crosses threshold. + // Note: the curator refreshes `distinct_sources` from the entity + // index during recompute, so we also need enough `query_hits_30d` + // to keep hotness above `TOPIC_CREATION_THRESHOLD` once the index + // is queried (two indexed sources below → distinct_sources → 2). + let mut counters = + crate::openhuman::memory::tree::topic_tree::types::HotnessCounters::fresh(entity_id, 0); + counters.mention_count_30d = 1_000; + counters.distinct_sources = 2; + counters.last_seen_ms = Some(Utc::now().timestamp_millis()); + counters.query_hits_30d = 5; + counters.ingests_since_check = + crate::openhuman::memory::tree::topic_tree::types::TOPIC_RECHECK_EVERY - 1; + crate::openhuman::memory::tree::topic_tree::store::upsert(&cfg, &counters).unwrap(); + + // Seed a leaf in slack and gmail referencing Alice. + let c1 = persist_chunk(&cfg, "slack:#eng", 0, 1_700_000_000_000, 100); + let c2 = persist_chunk(&cfg, "gmail:alice", 0, 1_700_000_010_000, 100); + let e = CanonicalEntity { + canonical_id: entity_id.into(), + kind: EntityKind::Email, + surface: entity_id.into(), + span_start: 0, + span_end: entity_id.len() as u32, + score: 1.0, + }; + index_entity(&cfg, &e, &c1, "leaf", 1_700_000_000_000, Some("slack:#eng")).unwrap(); + index_entity( + &cfg, + &e, + &c2, + "leaf", + 1_700_000_010_000, + Some("gmail:alice"), + ) + .unwrap(); + + // A third leaf arrives — should both fan out to (future) topic tree + // and push the curator over the recheck cadence, materialising it. + let c3 = persist_chunk(&cfg, "slack:#eng", 1, 1_700_000_020_000, 100); + let leaf = LeafRef { + chunk_id: c3.clone(), + token_count: 100, + timestamp: Utc.timestamp_millis_opt(1_700_000_020_000).unwrap(), + content: "new mention".into(), + entities: vec![entity_id.into()], + topics: vec![], + score: 0.5, + }; + + route_leaf_to_topic_trees(&cfg, &leaf, &[entity_id.to_string()], &summariser) + .await + .unwrap(); + + // Topic tree now exists. + let tree = src_store::get_tree_by_scope(&cfg, TreeKind::Topic, entity_id) + .unwrap() + .expect("topic tree should be materialised"); + assert_eq!(tree.kind, TreeKind::Topic); + assert_eq!(tree.scope, entity_id); + // Backfill pulled c1 + c2 into the buffer. (c3 didn't get into the + // entity index during this test since we didn't run the full ingest + // path — we're exercising routing in isolation.) + let buf = src_store::get_buffer(&cfg, &tree.id, 0).unwrap(); + assert!( + buf.item_ids.len() >= 2, + "backfill should pull historic leaves" + ); + } +} diff --git a/src/openhuman/memory/tree/topic_tree/store.rs b/src/openhuman/memory/tree/topic_tree/store.rs new file mode 100644 index 000000000..e7c54a7f0 --- /dev/null +++ b/src/openhuman/memory/tree/topic_tree/store.rs @@ -0,0 +1,245 @@ +//! SQLite persistence for topic-tree-specific state (#709 Phase 3c). +//! +//! The only new table owned here is `mem_tree_entity_hotness` — the +//! per-entity counter block driving lazy materialisation. Tree rows and +//! summary nodes are reused from [`super::super::source_tree::store`] via +//! the shared `mem_tree_trees` / `mem_tree_summaries` / `mem_tree_buffers` +//! tables, which already carry a `kind` column that discriminates +//! `source` from `topic`. No schema additions for those tables in Phase +//! 3c — only the new hotness table. +//! +//! Schema for `mem_tree_entity_hotness` is declared in +//! [`super::super::store::SCHEMA`] (the sibling Phase 1 store file) so +//! migrations all run through the same `with_connection` entry point. + +use anyhow::{Context, Result}; +use chrono::Utc; +use rusqlite::{params, OptionalExtension}; + +use crate::openhuman::config::Config; +use crate::openhuman::memory::tree::store::with_connection; +use crate::openhuman::memory::tree::topic_tree::types::HotnessCounters; + +/// Fetch the hotness row for `entity_id`, or `None` if the entity has +/// never been seen. Callers usually want [`get_or_fresh`] instead. +pub fn get(config: &Config, entity_id: &str) -> Result> { + with_connection(config, |conn| { + let mut stmt = conn.prepare( + "SELECT entity_id, mention_count_30d, distinct_sources, last_seen_ms, + query_hits_30d, graph_centrality, ingests_since_check, + last_hotness, last_updated_ms + FROM mem_tree_entity_hotness WHERE entity_id = ?1", + )?; + let row = stmt + .query_row(params![entity_id], row_to_counters) + .optional() + .context("failed to query mem_tree_entity_hotness")?; + Ok(row) + }) +} + +/// Fetch the hotness row, or return a fresh (all-zero) row if the entity +/// has never been seen. The fresh row is NOT persisted — callers must +/// [`upsert`] it explicitly after bumping counters. +pub fn get_or_fresh(config: &Config, entity_id: &str) -> Result { + match get(config, entity_id)? { + Some(c) => Ok(c), + None => Ok(HotnessCounters::fresh( + entity_id, + Utc::now().timestamp_millis(), + )), + } +} + +/// Upsert the full counter row. Idempotent on `entity_id`. +pub fn upsert(config: &Config, counters: &HotnessCounters) -> Result<()> { + with_connection(config, |conn| { + conn.execute( + "INSERT INTO mem_tree_entity_hotness ( + entity_id, mention_count_30d, distinct_sources, last_seen_ms, + query_hits_30d, graph_centrality, ingests_since_check, + last_hotness, last_updated_ms + ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9) + ON CONFLICT(entity_id) DO UPDATE SET + mention_count_30d = excluded.mention_count_30d, + distinct_sources = excluded.distinct_sources, + last_seen_ms = excluded.last_seen_ms, + query_hits_30d = excluded.query_hits_30d, + graph_centrality = excluded.graph_centrality, + ingests_since_check = excluded.ingests_since_check, + last_hotness = excluded.last_hotness, + last_updated_ms = excluded.last_updated_ms", + params![ + counters.entity_id, + counters.mention_count_30d, + counters.distinct_sources, + counters.last_seen_ms, + counters.query_hits_30d, + counters.graph_centrality, + counters.ingests_since_check, + counters.last_hotness, + counters.last_updated_ms, + ], + ) + .with_context(|| { + format!( + "failed to upsert mem_tree_entity_hotness for {}", + counters.entity_id + ) + })?; + Ok(()) + }) +} + +/// Count `(node_id) → DISTINCT tree_id` in the entity index for `entity_id`. +/// Used by the curator to refresh `distinct_sources` during the periodic +/// hotness recompute without rescanning every chunk. +pub fn distinct_sources_for(config: &Config, entity_id: &str) -> Result { + with_connection(config, |conn| { + let n: i64 = conn + .query_row( + "SELECT COUNT(DISTINCT tree_id) + FROM mem_tree_entity_index + WHERE entity_id = ?1 AND tree_id IS NOT NULL", + params![entity_id], + |r| r.get(0), + ) + .context("failed to count distinct sources")?; + Ok(n.max(0) as u32) + }) +} + +/// Test / diagnostic helper. +pub fn count(config: &Config) -> Result { + with_connection(config, |conn| { + let n: i64 = conn + .query_row("SELECT COUNT(*) FROM mem_tree_entity_hotness", [], |r| { + r.get(0) + }) + .context("failed to count mem_tree_entity_hotness")?; + Ok(n.max(0) as u64) + }) +} + +fn row_to_counters(row: &rusqlite::Row<'_>) -> rusqlite::Result { + Ok(HotnessCounters { + entity_id: row.get(0)?, + mention_count_30d: row.get::<_, i64>(1)?.max(0) as u32, + distinct_sources: row.get::<_, i64>(2)?.max(0) as u32, + last_seen_ms: row.get(3)?, + query_hits_30d: row.get::<_, i64>(4)?.max(0) as u32, + graph_centrality: row.get(5)?, + ingests_since_check: row.get::<_, i64>(6)?.max(0) as u32, + last_hotness: row.get(7)?, + last_updated_ms: row.get(8)?, + }) +} + +#[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_missing_is_none() { + let (_tmp, cfg) = test_config(); + assert!(get(&cfg, "email:alice@example.com").unwrap().is_none()); + } + + #[test] + fn get_or_fresh_returns_zero_row() { + let (_tmp, cfg) = test_config(); + let c = get_or_fresh(&cfg, "email:alice@example.com").unwrap(); + assert_eq!(c.entity_id, "email:alice@example.com"); + assert_eq!(c.mention_count_30d, 0); + assert_eq!(c.distinct_sources, 0); + assert!(c.last_hotness.is_none()); + // Not persisted — still zero rows in the table. + assert_eq!(count(&cfg).unwrap(), 0); + } + + #[test] + fn upsert_round_trip() { + let (_tmp, cfg) = test_config(); + let c = HotnessCounters { + entity_id: "email:alice@example.com".into(), + mention_count_30d: 12, + distinct_sources: 3, + last_seen_ms: Some(1_700_000_000_000), + query_hits_30d: 2, + graph_centrality: Some(0.25), + ingests_since_check: 40, + last_hotness: Some(9.5), + last_updated_ms: 1_700_000_123_000, + }; + upsert(&cfg, &c).unwrap(); + let got = get(&cfg, &c.entity_id).unwrap().unwrap(); + assert_eq!(got, c); + assert_eq!(count(&cfg).unwrap(), 1); + } + + #[test] + fn upsert_is_idempotent_and_updates_fields() { + let (_tmp, cfg) = test_config(); + let mut c = HotnessCounters::fresh("email:alice@example.com", 0); + c.mention_count_30d = 1; + upsert(&cfg, &c).unwrap(); + c.mention_count_30d = 99; + c.last_updated_ms = 500; + upsert(&cfg, &c).unwrap(); + assert_eq!(count(&cfg).unwrap(), 1); + let got = get(&cfg, "email:alice@example.com").unwrap().unwrap(); + assert_eq!(got.mention_count_30d, 99); + assert_eq!(got.last_updated_ms, 500); + } + + #[test] + fn distinct_sources_counts_trees() { + use crate::openhuman::memory::tree::score::extract::EntityKind; + use crate::openhuman::memory::tree::score::resolver::CanonicalEntity; + use crate::openhuman::memory::tree::score::store::index_entity; + let (_tmp, cfg) = test_config(); + let e = CanonicalEntity { + canonical_id: "email:alice@example.com".into(), + kind: EntityKind::Email, + surface: "alice@example.com".into(), + span_start: 0, + span_end: 17, + score: 1.0, + }; + index_entity(&cfg, &e, "chunk-1", "leaf", 1000, Some("source:slack")).unwrap(); + index_entity(&cfg, &e, "chunk-2", "leaf", 2000, Some("source:gmail")).unwrap(); + index_entity(&cfg, &e, "chunk-3", "leaf", 3000, Some("source:slack")).unwrap(); + // 3 rows but only 2 distinct tree_ids. + let n = distinct_sources_for(&cfg, "email:alice@example.com").unwrap(); + assert_eq!(n, 2); + } + + #[test] + fn distinct_sources_ignores_null_tree_id() { + use crate::openhuman::memory::tree::score::extract::EntityKind; + use crate::openhuman::memory::tree::score::resolver::CanonicalEntity; + use crate::openhuman::memory::tree::score::store::index_entity; + let (_tmp, cfg) = test_config(); + let e = CanonicalEntity { + canonical_id: "email:alice@example.com".into(), + kind: EntityKind::Email, + surface: "alice@example.com".into(), + span_start: 0, + span_end: 17, + score: 1.0, + }; + // tree_id = None — should not count toward distinct_sources. + index_entity(&cfg, &e, "chunk-1", "leaf", 1000, None).unwrap(); + index_entity(&cfg, &e, "chunk-2", "leaf", 2000, Some("source:slack")).unwrap(); + let n = distinct_sources_for(&cfg, "email:alice@example.com").unwrap(); + assert_eq!(n, 1); + } +} diff --git a/src/openhuman/memory/tree/topic_tree/types.rs b/src/openhuman/memory/tree/topic_tree/types.rs new file mode 100644 index 000000000..734c96840 --- /dev/null +++ b/src/openhuman/memory/tree/topic_tree/types.rs @@ -0,0 +1,150 @@ +//! Core types for Phase 3c — lazy topic-tree materialisation (#709). +//! +//! A *topic tree* is a per-entity summary tree whose leaves are all chunks +//! mentioning that entity, regardless of the source they came from. They +//! are materialised lazily, driven by a cheap arithmetic *hotness* score +//! over pre-existing signals. Tree mechanics (buffer / seal / cascade) +//! reuse [`source_tree`] end-to-end — topic trees only differ by the +//! `TreeKind::Topic` discriminator and the per-entity `scope`. +//! +//! This file defines: +//! - [`EntityIndexStats`] — input record for the hotness calculation +//! - [`HotnessCounters`] — the persisted row in `mem_tree_entity_hotness` +//! - threshold / cadence constants ([`TOPIC_CREATION_THRESHOLD`], +//! [`TOPIC_ARCHIVE_THRESHOLD`], [`TOPIC_RECHECK_EVERY`]) +//! +//! Persistence helpers for these types live in [`super::store`]. + +use serde::{Deserialize, Serialize}; + +/// Hotness threshold above which a topic tree is materialised for an +/// entity. Tuned (per design) to roughly "several mentions across a few +/// sources" — see [`super::hotness::hotness`] for the formula. +pub const TOPIC_CREATION_THRESHOLD: f32 = 10.0; + +/// Hotness threshold below which a topic tree becomes an archive candidate. +/// Archiving is a primitive in Phase 3c — the scheduled sweep is deferred. +pub const TOPIC_ARCHIVE_THRESHOLD: f32 = 2.0; + +/// How often (in ingests touching the entity) to recompute hotness from +/// the full [`EntityIndexStats`]. Between recomputes we only bump +/// `mention_count_30d` and `last_seen_ms` — cheap integer arithmetic. +pub const TOPIC_RECHECK_EVERY: u32 = 100; + +/// Input record fed to [`super::hotness::hotness`]. +/// +/// Every field is a signal that already exists somewhere in the memory +/// tree (scoring rows, entity index, potential future graph metrics); the +/// struct is an explicit contract so the hotness math can be unit-tested +/// in isolation without touching SQLite. +#[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)] +pub struct EntityIndexStats { + /// Total mentions in the last 30 days. Phase 3c currently bumps this + /// forever — the 30d window is a TODO once we have a billable clock. + pub mention_count_30d: u32, + /// Number of distinct source trees this entity has appeared in. A + /// cross-source signal — an entity spoken about in one chat channel + /// but nowhere else is less interesting than one that appears in + /// Slack + email + docs. + pub distinct_sources: u32, + /// Epoch-millis of the last ingest that referenced this entity. + pub last_seen_ms: Option, + /// Reserved for Phase 4 retrieval: bump whenever a user query returns + /// this entity. Phase 3c stores the column but never increments it. + pub query_hits_30d: u32, + /// Reserved for a later phase: graph centrality from the entity graph. + /// `None` means "unknown" — not "zero". See [`super::hotness::hotness`]. + pub graph_centrality: Option, +} + +/// Row persisted in `mem_tree_entity_hotness`. Callers interact with this +/// through [`super::store`]; [`EntityIndexStats`] is the hotness-compute +/// view derived from it. +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] +pub struct HotnessCounters { + pub entity_id: String, + pub mention_count_30d: u32, + pub distinct_sources: u32, + pub last_seen_ms: Option, + pub query_hits_30d: u32, + pub graph_centrality: Option, + /// Counts ingests **touching this entity** since the last full hotness + /// recompute. When `>= TOPIC_RECHECK_EVERY` the curator refreshes + /// `distinct_sources` / `last_hotness` and resets this to 0. + pub ingests_since_check: u32, + pub last_hotness: Option, + pub last_updated_ms: i64, +} + +impl HotnessCounters { + /// Fresh row for an entity seen for the first time. + pub fn fresh(entity_id: &str, now_ms: i64) -> Self { + Self { + entity_id: entity_id.to_string(), + mention_count_30d: 0, + distinct_sources: 0, + last_seen_ms: None, + query_hits_30d: 0, + graph_centrality: None, + ingests_since_check: 0, + last_hotness: None, + last_updated_ms: now_ms, + } + } + + /// Project the persisted row into an [`EntityIndexStats`] ready for + /// [`super::hotness::hotness`]. + pub fn stats(&self) -> EntityIndexStats { + EntityIndexStats { + mention_count_30d: self.mention_count_30d, + distinct_sources: self.distinct_sources, + last_seen_ms: self.last_seen_ms, + query_hits_30d: self.query_hits_30d, + graph_centrality: self.graph_centrality, + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn fresh_counters_are_zero() { + let c = HotnessCounters::fresh("email:alice@example.com", 1_700_000_000_000); + assert_eq!(c.entity_id, "email:alice@example.com"); + assert_eq!(c.mention_count_30d, 0); + assert_eq!(c.distinct_sources, 0); + assert_eq!(c.ingests_since_check, 0); + assert!(c.last_hotness.is_none()); + assert!(c.last_seen_ms.is_none()); + assert_eq!(c.last_updated_ms, 1_700_000_000_000); + } + + #[test] + fn stats_projection_mirrors_row() { + let c = HotnessCounters { + entity_id: "e".into(), + mention_count_30d: 5, + distinct_sources: 2, + last_seen_ms: Some(42), + query_hits_30d: 1, + graph_centrality: Some(0.3), + ingests_since_check: 4, + last_hotness: Some(9.9), + last_updated_ms: 100, + }; + let s = c.stats(); + assert_eq!(s.mention_count_30d, 5); + assert_eq!(s.distinct_sources, 2); + assert_eq!(s.last_seen_ms, Some(42)); + assert_eq!(s.query_hits_30d, 1); + assert_eq!(s.graph_centrality, Some(0.3)); + } + + #[test] + fn thresholds_make_creation_strictly_above_archive() { + assert!(TOPIC_CREATION_THRESHOLD > TOPIC_ARCHIVE_THRESHOLD); + assert!(TOPIC_RECHECK_EVERY > 0); + } +}