admin/openhuman
1
0
Fork 0
mirror of https://github.com/tinyhumansai/openhuman.git synced 2026-07-27 21:08:00 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(memory): topic trees (lazy hotness-driven materialisation) (#709) (#799)

Browse Source 
Phase 3c of the memory architecture (umbrella #711). Adds per-entity
topic trees spawned lazily when an entity's hotness crosses a threshold,
so Phase 4 retrieval can resolve "what did Alice say about Phoenix?"
without scanning the full chunk pool.

## What's in this commit

New module `src/openhuman/memory/tree/topic_tree/`:

- `types.rs` — `EntityIndexStats` (hotness input), `HotnessCounters` (the
  persisted row) and the `TOPIC_CREATION_THRESHOLD=10.0`,
  `TOPIC_ARCHIVE_THRESHOLD=2.0`, `TOPIC_RECHECK_EVERY=100` constants.
- `hotness.rs` — pure arithmetic scorer plus deterministic
  `hotness_at(entity_id, stats, now_ms)` variant for tests, and a
  piecewise-linear `recency_decay` helper.
- `store.rs` — SQLite helpers (`get`, `get_or_fresh`, `upsert`,
  `distinct_sources_for`, `count`) for the new `mem_tree_entity_hotness`
  table.
- `registry.rs` — `get_or_create_topic_tree`, `force_create_topic_tree`,
  `list_topic_trees`, `archive_topic_tree`. Race-recovers on UNIQUE
  violations via `is_unique_violation`, mirroring `source_tree::registry`.
- `curator.rs` — `maybe_spawn_topic_tree` (per-ingest tick) and
  `force_recompute` (admin path). Implements the
  `TOPIC_RECHECK_EVERY`-gated recompute: refresh `distinct_sources` from
  the entity index, compute hotness, spawn + backfill if threshold
  crossed.
- `backfill.rs` — `backfill_topic_tree` walks the entity index and routes
  every historic leaf (ts-ASC ordered) through
  `source_tree::bucket_seal::append_leaf`. Capped at 500 leaves. Skips
  summary-node hits so topic trees only ever ingest raw leaves. Missing
  chunks log a warn and are skipped, never failing the spawn.
- `routing.rs` — `route_leaf_to_topic_trees` fans a kept leaf out to
  every active matching topic tree and ticks the curator for each
  entity. Archived trees are skipped but counters still bump.

Wired from ingest (`tree/ingest.rs::append_leaves_to_tree`) AFTER the
source-tree `append_leaf` succeeds — non-fatal on error, logged at warn
level so routing issues never poison the ingest hot path.

## Hotness scoring (pure)

```
hotness = ln(mentions + 1)          // dampened volume
        + 0.5 * distinct_sources    // cross-source bonus
        + recency_decay(last_seen)  // 1.0 @ day 0 → 0 @ day 30
        + graph_centrality          // Phase 4+; None → 0
        + 2.0 * query_hits          // retrieval feedback; Phase 4+
```

`graph_centrality` and `query_hits_30d` columns are persisted but the
increment code paths are deferred to later phases, as planned.

## Schema (additive, idempotent)

New table `mem_tree_entity_hotness` (keyed on `entity_id`) added to the
Phase 1 `SCHEMA` constant in `tree/store.rs` so it migrates through the
same `with_connection` path as the existing tables. `CREATE TABLE IF NOT
EXISTS` keeps it re-run-safe. An ancillary index on `last_hotness`
supports future sweep queries.

## Routing

The ingest path is the only non-admin caller. After `append_leaf` puts a
leaf in the source tree, `route_leaf_to_topic_trees` runs with the
chunk's canonical entity list. For each entity:

1. If an active topic tree exists, append the leaf to it (reusing
   `source_tree::bucket_seal::append_leaf` with `entities=[entity_id]`).
2. Tick `maybe_spawn_topic_tree` — may bump counters or, on cadence,
   recompute hotness and spawn + backfill a new tree.

Per-entity errors are caught and logged; a top-level failure is demoted
to a warn in the ingest caller so the source-tree append always wins.

## Reuses from Phase 3a

- `source_tree::bucket_seal::append_leaf` — same `&Tree` API works for
  `TreeKind::Topic` end-to-end.
- `source_tree::summariser::{Summariser, InertSummariser}` — honest stub
  emits empty entity/topic vecs on summary nodes.
- `mem_tree_trees` / `mem_tree_summaries` / `mem_tree_buffers` schema,
  discriminated by `kind = 'topic'` and `scope = <entity canonical id>`.
- Registry race recovery via `is_unique_violation` (catches UNIQUE on
  `insert_tree`, re-queries on collision).
- Idempotent append (duplicate `item_ids` in the L0 buffer are no-ops),
  so backfill is safe to re-run.

## Tests

40 new tests (10 hotness, 3 types, 6 store, 7 registry, 4 curator, 4
backfill, 5 routing, 1 integration). `memory::tree` suite: 201 passing,
0 regressions (Phase 3a baseline was 161 → +40 new).

Coverage highlights:
- Hotness pure math (zero-entity, spike-over-threshold, old-but-widely-
  cited retains signal, query-hit boost, recency decay edges).
- Curator: first-ingest-just-bumps, no-spawn-below-threshold,
  spawn-fires-exactly-once-when-crossed, cadence gating.
- Backfill: appends-all-entity-leaves, skips missing chunks,
  idempotent, skips summary nodes.
- Routing: empty-entities-noop, appends-to-existing-tree, archived-tree-
  skipped, multi-entity-fan-out, end-to-end-integration-materialisation.
- Registry: idempotent get-or-create, UNIQUE race recovery, archive
  flips status (not deletion), kind/scope cleanly separated from source.

## Deliberate non-goals (deferred)

- No JSON-RPC surface (out of scope for Phase 3c core).
- No archive cron sweep — `archive_topic_tree` primitive only.
- `graph_centrality` / `query_hits_30d` increments deferred to later
  phases (columns exist, reads work, writes are Phase 4+).

## Stacked on #789

Base is `feat/709-summary-trees`. Merge #789 first, then rebase this
branch.

Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
sanil-23
2026-04-22 13:40:08 -07:00
committed by GitHub
co-authored by Claude Opus 4.7
parent 48897af9b3
commit 32b5afd1d9
11 changed files with 1991 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/openhuman/memory/tree/ingest.rs
+14 -2
View File
@@ -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::<Vec<_>>(),
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(())
}
src/openhuman/memory/tree/mod.rs
+1
View File
@@ -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::{
src/openhuman/memory/tree/store.rs
+21
View File
@@ -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.
src/openhuman/memory/tree/topic_tree/backfill.rs
+304
View File
@@ -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<usize> {
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);
}
}
src/openhuman/memory/tree/topic_tree/curator.rs
+370
View File
@@ -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<SpawnOutcome> {
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<SpawnOutcome> {
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<SpawnOutcome> {
// 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<Option<Tree>> {
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:?}"),
}
}
}
src/openhuman/memory/tree/topic_tree/hotness.rs
+203
View File
@@ -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<i64>, 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<i64>) -> 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);
}
}
src/openhuman/memory/tree/topic_tree/mod.rs
+46
View File
@@ -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,
};
src/openhuman/memory/tree/topic_tree/registry.rs
+289
View File
@@ -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 = <entity canonical id>`.
//! 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<Tree> {
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<Tree> {
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<Vec<Tree>> {
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::<rusqlite::Result<Vec<_>>>()
.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<Tree> {
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::<rusqlite::Error>() {
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<Tree> {
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<String> = 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<i64> = 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);
}
}
}
src/openhuman/memory/tree/topic_tree/routing.rs
+348
View File
@@ -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"
);
}
}
src/openhuman/memory/tree/topic_tree/store.rs
+245
View File
@@ -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<Option<HotnessCounters>> {
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<HotnessCounters> {
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<u32> {
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<u64> {
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<HotnessCounters> {
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);
}
}
src/openhuman/memory/tree/topic_tree/types.rs
+150
View File
@@ -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<i64>,
/// 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<f32>,
}
/// 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<i64>,
pub query_hits_30d: u32,
pub graph_centrality: Option<f32>,
/// 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<f32>,
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);
}
}