feat(memory): source trees + bucket-seal foundation (#709)

Phase 3a of the memory architecture (#711 umbrella). Lifts admitted leaves
into a per-source hierarchy so Phase 4 retrieval has a tree to walk.

## What's in this PR

* **Schema** — three new tables in `chunks.db` alongside Phase 1/2:
  `mem_tree_trees`, `mem_tree_summaries`, `mem_tree_buffers`. Plus a
  `parent_summary_id` backlink column on `mem_tree_chunks`. All migrations
  are additive and idempotent (`ALTER TABLE ADD COLUMN`, same pattern as
  Phase 2).
* **`source_tree/` module** — isolated subdir; does not touch the legacy
  `openhuman::memory` layer.
  - `types.rs` — `Tree`, `SummaryNode`, `Buffer`, `TreeKind` (Source/Topic/Global),
    `TreeStatus`. Constants: `TOKEN_BUDGET = 10_000`, `DEFAULT_FLUSH_AGE_SECS = 7d`.
  - `store.rs` — CRUD for all three tables via the shared `with_connection`
    entry point. Writes are transactional; summary insert is idempotent
    on primary key (INSERT OR IGNORE).
  - `registry.rs` — `get_or_create_source_tree(scope)` — idempotent lookup
    keyed on `(kind, scope)`.
  - `bucket_seal.rs` — `append_leaf` + cascade: push a leaf into L0, seal
    when `token_sum >= TOKEN_BUDGET`, recurse upward. One seal = one
    transaction; children get a parent backlink (leaves via
    `parent_summary_id`, summaries via `parent_id`). Tree's `max_level`
    and `root_id` bump on root split.
  - `flush.rs` — `flush_stale_buffers(max_age)` force-seals any buffer
    whose `oldest_at` is older than `max_age`. Primitive only; wiring
    into a scheduler is out of scope.
  - `summariser/` — `Summariser` trait + `InertSummariser` fallback
    (concatenate children, union entities preserving first-seen order,
    hard-truncate to budget). Trait is async; Ollama implementation slots
    in later without breaking callers.
* **Ingest wiring** — `tree/ingest.rs::persist` now calls `append_leaf`
  once per kept chunk after chunks + scores land. Failures are logged at
  warn level and do not fail the ingest — leaves are already persisted
  and the next flush can still rebuild the tree.

## Concurrency / LLM

The async summariser call happens OUTSIDE any DB transaction, so a slow
LLM never holds SQLite locks. Blocking DB calls inside `append_leaf` are
acceptable for Phase 3a because the Inert summariser does no real I/O;
when a networked summariser lands, wrap the DB calls in
`tokio::task::spawn_blocking`.

## Tests (23 new, all green)

* `types` — enum round-trips, buffer staleness predicates
* `store` — tree insert + unique scope, summary insert + idempotence,
  buffer upsert/clear, stale-buffer ordering
* `registry` — `get_or_create` idempotence, distinct scopes yield distinct
  trees, id prefix format
* `summariser/inert` — provenance-prefixed concat, entity union with
  order preservation, budget truncation, empty-content skip
* `bucket_seal` — append-below-budget is buffered only (no seal);
  12k-token cross-over produces an L1 summary with correct `child_ids`,
  L0 cleared, L1 carries the new summary id, tree `max_level`/`root_id`
  updated, leaf → parent backlink populated
* `flush` — stale buffer force-seals under budget; recent buffer is a no-op

Broader `memory::tree` suite: 160 tests pass (23 new + 137 existing Phase
1/2), no regressions.

## Stacked on #775

This PR applies on top of `feat/708-memory-llm-ner` (PR #775, Phase 2
LLM-NER follow-up). Merge #775 first; this branch will rebase cleanly
onto main.

## Out of scope (tracked separately)

* **3b — Global activity digest tree** — time-aligned cross-source recap
* **3c — Topic trees** — hotness-driven per-entity materialisation
* **Phase 4 (#710)** — retrieval / query / gate

Closes part of #709 (source-tree foundation).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

src/openhuman/memory/tree/ingest.rs

@@ -20,6 +20,9 @@ use crate::openhuman::memory::tree::canonicalize::{
 };
 use crate::openhuman::memory::tree::chunker::{chunk_markdown, ChunkerInput, ChunkerOptions};
 use crate::openhuman::memory::tree::score::{self, ScoreResult, ScoringConfig};
+use crate::openhuman::memory::tree::source_tree::{
+    append_leaf, get_or_create_source_tree, InertSummariser, LeafRef,
+};
 use crate::openhuman::memory::tree::store;
 use crate::openhuman::memory::tree::types::Chunk;
 
@@ -182,6 +185,20 @@ async fn persist(
     .await
     .map_err(|e| anyhow::anyhow!("persist join error: {e}"))??;
 
+    // 5. Source-tree append (Phase 3a #709). Each kept leaf pushes into
+    //    the tree's L0 buffer and cascades upward when token_sum crosses
+    //    the budget. Entities/topics from the scorer are threaded in so
+    //    sealed summaries inherit the child signal set. Failures here
+    //    log at warn level but don't fail the ingest — leaves are already
+    //    persisted, and a later flush/retry can still rebuild the tree.
+    if let Err(e) = append_leaves_to_tree(config, source_id, &kept_chunks, &all_results).await {
+        log::warn!(
+            "[memory_tree::ingest] source_tree append failed source_id={} err={:#}",
+            source_id,
+            e
+        );
+    }
+
     Ok(IngestResult {
         source_id: source_id.to_string(),
         chunks_written: written,
@@ -190,6 +207,55 @@ async fn persist(
     })
 }
 
+/// Push every kept chunk into its source tree. Scoped to Phase 3a — all
+/// chunks from one ingest batch share the same `source_id`, so they share
+/// one tree lookup. The inert summariser is the default; future wiring
+/// can swap in an LLM summariser here.
+async fn append_leaves_to_tree(
+    config: &Config,
+    source_id: &str,
+    kept_chunks: &[Chunk],
+    all_results: &[(ScoreResult, i64)],
+) -> Result<()> {
+    if kept_chunks.is_empty() {
+        return Ok(());
+    }
+    let tree = get_or_create_source_tree(config, source_id)?;
+    let summariser = InertSummariser::new();
+
+    // Build a chunk_id → (score, entities, topics) map for quick lookup.
+    use std::collections::HashMap;
+    let mut score_by_id: HashMap<String, &ScoreResult> = HashMap::new();
+    for (r, _) in all_results {
+        score_by_id.insert(r.chunk_id.clone(), r);
+    }
+
+    for chunk in kept_chunks {
+        let (score_value, entities, topics) = match score_by_id.get(&chunk.id) {
+            Some(r) => (
+                r.total,
+                r.canonical_entities
+                    .iter()
+                    .map(|e| e.canonical_id.clone())
+                    .collect(),
+                chunk.metadata.tags.clone(),
+            ),
+            None => (0.0, Vec::new(), chunk.metadata.tags.clone()),
+        };
+        let leaf = LeafRef {
+            chunk_id: chunk.id.clone(),
+            token_count: chunk.token_count,
+            timestamp: chunk.metadata.timestamp,
+            content: chunk.content.clone(),
+            entities,
+            topics,
+            score: score_value,
+        };
+        append_leaf(config, &tree, &leaf, &summariser).await?;
+    }
+    Ok(())
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

src/openhuman/memory/tree/mod.rs

@@ -28,6 +28,7 @@ pub mod ingest;
 pub mod rpc;
 pub mod schemas;
 pub mod score;
+pub mod source_tree;
 pub mod store;
 pub mod types;
 

src/openhuman/memory/tree/source_tree/bucket_seal.rs

@@ -0,0 +1,545 @@
+//! Append + cascade-seal for summary trees (#709).
+//!
+//! `append_leaf` pushes a persisted chunk into the L0 buffer of a tree. If
+//! the buffer's running `token_sum` crosses `TOKEN_BUDGET`, the buffer
+//! seals into a level-1 summary, its items move into the summary's
+//! `child_ids`, the buffer clears, and the new summary id is queued at
+//! level 2. The cascade continues upward until a buffer stays under the
+//! token budget.
+//!
+//! Concurrency: Phase 3a assumes a single-process SQLite workspace. All
+//! writes in one seal step run in a single transaction; the async
+//! summariser call happens outside any open transaction so a slow LLM
+//! doesn't hold DB locks. Callers should serialise `append_leaf` per
+//! tree — ingest achieves this by processing one batch's chunks
+//! sequentially inside its `persist` task. Blocking SQLite calls inside
+//! this async function are acceptable for Phase 3a because the Inert
+//! summariser does no real I/O; when a networked summariser lands, wrap
+//! DB calls in `tokio::task::spawn_blocking` to keep the runtime healthy.
+
+use anyhow::{Context, Result};
+use chrono::{DateTime, Utc};
+use rusqlite::Transaction;
+
+use crate::openhuman::config::Config;
+use crate::openhuman::memory::tree::source_tree::registry::new_summary_id;
+use crate::openhuman::memory::tree::source_tree::store;
+use crate::openhuman::memory::tree::source_tree::summariser::{
+    SummaryContext, SummaryInput, Summariser,
+};
+use crate::openhuman::memory::tree::source_tree::types::{
+    Buffer, SummaryNode, Tree, TreeKind, TOKEN_BUDGET,
+};
+use crate::openhuman::memory::tree::store::with_connection;
+
+/// Hard cap on cascade depth — prevents runaway loops if token accounting
+/// ever slips. 32 levels at even a 2x fan-in is more than enough for any
+/// realistic source.
+const MAX_CASCADE_DEPTH: u32 = 32;
+
+/// A single leaf being appended to an L0 buffer.
+#[derive(Clone, Debug)]
+pub struct LeafRef {
+    pub chunk_id: String,
+    pub token_count: u32,
+    pub timestamp: DateTime<Utc>,
+    pub content: String,
+    pub entities: Vec<String>,
+    pub topics: Vec<String>,
+    pub score: f32,
+}
+
+/// Append a leaf to the source tree for `tree`, sealing buffers as they
+/// fill. Returns the ids of any summaries that sealed during this call.
+pub async fn append_leaf(
+    config: &Config,
+    tree: &Tree,
+    leaf: &LeafRef,
+    summariser: &dyn Summariser,
+) -> Result<Vec<String>> {
+    log::debug!(
+        "[source_tree::bucket_seal] append_leaf tree_id={} leaf_id={} tokens={}",
+        tree.id,
+        leaf.chunk_id,
+        leaf.token_count
+    );
+
+    // 1. Push leaf into L0 buffer (transactional).
+    append_to_buffer(
+        config,
+        &tree.id,
+        0,
+        &leaf.chunk_id,
+        leaf.token_count as i64,
+        leaf.timestamp,
+    )?;
+
+    // 2. Cascade seals upward until a level stays under budget.
+    cascade_seals(config, tree, summariser).await
+}
+
+/// Transactionally append a single item to `(tree_id, level)`'s buffer.
+fn append_to_buffer(
+    config: &Config,
+    tree_id: &str,
+    level: u32,
+    item_id: &str,
+    token_delta: i64,
+    item_ts: DateTime<Utc>,
+) -> Result<()> {
+    with_connection(config, |conn| {
+        let tx = conn.unchecked_transaction()?;
+        let mut buf = store::get_buffer_conn(&tx, tree_id, level)?;
+        buf.item_ids.push(item_id.to_string());
+        buf.token_sum = buf.token_sum.saturating_add(token_delta);
+        buf.oldest_at = match buf.oldest_at {
+            Some(existing) => Some(existing.min(item_ts)),
+            None => Some(item_ts),
+        };
+        store::upsert_buffer_tx(&tx, &buf)?;
+        tx.commit()?;
+        Ok(())
+    })
+}
+
+async fn cascade_seals(
+    config: &Config,
+    tree: &Tree,
+    summariser: &dyn Summariser,
+) -> Result<Vec<String>> {
+    cascade_all_from(config, tree, 0, summariser, None).await
+}
+
+/// Seal buffers starting at `start_level` and cascade upward. When
+/// `force_now` is `Some`, the buffer at `start_level` is sealed regardless
+/// of token budget (used by time-based flush). Upper levels are sealed
+/// only when they cross the budget.
+pub async fn cascade_all_from(
+    config: &Config,
+    tree: &Tree,
+    start_level: u32,
+    summariser: &dyn Summariser,
+    force_now: Option<DateTime<Utc>>,
+) -> Result<Vec<String>> {
+    let mut sealed_ids: Vec<String> = Vec::new();
+    let mut level: u32 = start_level;
+    let mut first_iteration = true;
+
+    for _ in 0..MAX_CASCADE_DEPTH {
+        let buf = store::get_buffer(config, &tree.id, level)?;
+        let forced = first_iteration && force_now.is_some();
+        first_iteration = false;
+
+        if !forced && !should_seal(&buf) {
+            log::debug!(
+                "[source_tree::bucket_seal] cascade done tree_id={} stop_level={} token_sum={}",
+                tree.id,
+                level,
+                buf.token_sum
+            );
+            break;
+        }
+        if buf.is_empty() {
+            log::debug!(
+                "[source_tree::bucket_seal] cascade hit empty buffer tree_id={} level={} — stopping",
+                tree.id,
+                level
+            );
+            break;
+        }
+
+        let summary_id = seal_one_level(config, tree, &buf, summariser).await?;
+        sealed_ids.push(summary_id);
+        level += 1;
+    }
+
+    Ok(sealed_ids)
+}
+
+fn should_seal(buf: &Buffer) -> bool {
+    !buf.is_empty() && buf.token_sum >= TOKEN_BUDGET as i64
+}
+
+/// Seal `buf` at `level` into one summary at `level + 1`. Returns the new
+/// summary id.
+async fn seal_one_level(
+    config: &Config,
+    tree: &Tree,
+    buf: &Buffer,
+    summariser: &dyn Summariser,
+) -> Result<String> {
+    let level = buf.level;
+    let target_level = level + 1;
+
+    // Hydrate inputs (synchronous DB reads).
+    let inputs = hydrate_inputs(config, level, &buf.item_ids)?;
+    if inputs.is_empty() {
+        anyhow::bail!(
+            "[source_tree::bucket_seal] refused to seal empty buffer tree_id={} level={}",
+            tree.id,
+            level
+        );
+    }
+
+    // Compute envelope across children (time range, max score).
+    let time_range_start = inputs
+        .iter()
+        .map(|i| i.time_range_start)
+        .min()
+        .unwrap_or_else(Utc::now);
+    let time_range_end = inputs
+        .iter()
+        .map(|i| i.time_range_end)
+        .max()
+        .unwrap_or_else(Utc::now);
+    let score = inputs
+        .iter()
+        .map(|i| i.score)
+        .fold(f32::NEG_INFINITY, f32::max)
+        .max(0.0);
+
+    // Run summariser — async, OUTSIDE any DB transaction.
+    let ctx = SummaryContext {
+        tree_id: &tree.id,
+        tree_kind: TreeKind::Source,
+        target_level,
+        token_budget: TOKEN_BUDGET,
+    };
+    let output = summariser
+        .summarise(&inputs, &ctx)
+        .await
+        .context("summariser failed during seal")?;
+
+    // Build the new summary node.
+    let now = Utc::now();
+    let summary_id = new_summary_id(target_level);
+    let node = SummaryNode {
+        id: summary_id.clone(),
+        tree_id: tree.id.clone(),
+        tree_kind: TreeKind::Source,
+        level: target_level,
+        parent_id: None,
+        child_ids: buf.item_ids.clone(),
+        content: output.content,
+        token_count: output.token_count,
+        entities: output.entities,
+        topics: output.topics,
+        time_range_start,
+        time_range_end,
+        score,
+        sealed_at: now,
+        deleted: false,
+    };
+
+    // Single write transaction: insert summary, clear this buffer, append
+    // summary id to parent buffer, bump tree max_level/root if needed.
+    // Re-read `max_level` from inside the tx so cascading seals within
+    // one call see the updated value from earlier levels.
+    let summary_id_for_closure = summary_id.clone();
+    let target_level_for_closure = target_level;
+    let tree_id = tree.id.clone();
+    with_connection(config, move |conn| {
+        let tx = conn.unchecked_transaction()?;
+
+        let current_max: u32 = tx
+            .query_row(
+                "SELECT max_level FROM mem_tree_trees WHERE id = ?1",
+                rusqlite::params![&tree_id],
+                |r| r.get::<_, i64>(0),
+            )
+            .map(|n| n.max(0) as u32)
+            .context("Failed to read current max_level for tree")?;
+
+        store::insert_summary_tx(&tx, &node)?;
+        // Backlink children → new parent so leaf/parent traversal is a
+        // single-row lookup in Phase 4. Skipped for levels already bound
+        // to a parent (shouldn't happen — a child seals at most once).
+        for child_id in &node.child_ids {
+            if level == 0 {
+                tx.execute(
+                    "UPDATE mem_tree_chunks
+                        SET parent_summary_id = ?1
+                      WHERE id = ?2 AND parent_summary_id IS NULL",
+                    rusqlite::params![&summary_id_for_closure, child_id],
+                )
+                .context("Failed to backlink chunk to parent summary")?;
+            } else {
+                tx.execute(
+                    "UPDATE mem_tree_summaries
+                        SET parent_id = ?1
+                      WHERE id = ?2 AND parent_id IS NULL",
+                    rusqlite::params![&summary_id_for_closure, child_id],
+                )
+                .context("Failed to backlink summary to parent summary")?;
+            }
+        }
+        store::clear_buffer_tx(&tx, &tree_id, level)?;
+
+        // Append to parent buffer.
+        let mut parent = store::get_buffer_conn(&tx, &tree_id, target_level_for_closure)?;
+        parent.item_ids.push(summary_id_for_closure.clone());
+        parent.token_sum = parent.token_sum.saturating_add(node.token_count as i64);
+        parent.oldest_at = match parent.oldest_at {
+            Some(existing) => Some(existing.min(time_range_start)),
+            None => Some(time_range_start),
+        };
+        store::upsert_buffer_tx(&tx, &parent)?;
+
+        // Update tree root / max_level if we just climbed.
+        if target_level_for_closure > current_max {
+            store::update_tree_after_seal_tx(
+                &tx,
+                &tree_id,
+                &summary_id_for_closure,
+                target_level_for_closure,
+                now,
+            )?;
+        } else {
+            // Same max level — still refresh last_sealed_at via same helper
+            // but keep existing root intact. Root tracking at the same
+            // level is resolved in Phase 4 retrieval.
+            refresh_last_sealed_tx(&tx, &tree_id, now)?;
+        }
+
+        tx.commit()?;
+        Ok(())
+    })?;
+
+    log::info!(
+        "[source_tree::bucket_seal] sealed tree_id={} level={}→{} summary_id={} children={}",
+        tree.id,
+        level,
+        target_level,
+        summary_id,
+        buf.item_ids.len()
+    );
+
+    Ok(summary_id)
+}
+
+fn refresh_last_sealed_tx(
+    tx: &Transaction<'_>,
+    tree_id: &str,
+    sealed_at: DateTime<Utc>,
+) -> Result<()> {
+    tx.execute(
+        "UPDATE mem_tree_trees SET last_sealed_at_ms = ?1 WHERE id = ?2",
+        rusqlite::params![sealed_at.timestamp_millis(), tree_id],
+    )
+    .with_context(|| format!("Failed to refresh last_sealed_at for tree {tree_id}"))?;
+    Ok(())
+}
+
+/// Fetch contributions for `item_ids`. At level 0 we pull from
+/// `mem_tree_chunks` + `mem_tree_score`; at ≥1 we pull from
+/// `mem_tree_summaries`.
+fn hydrate_inputs(
+    config: &Config,
+    level: u32,
+    item_ids: &[String],
+) -> Result<Vec<SummaryInput>> {
+    if level == 0 {
+        hydrate_leaf_inputs(config, item_ids)
+    } else {
+        hydrate_summary_inputs(config, item_ids)
+    }
+}
+
+fn hydrate_leaf_inputs(config: &Config, chunk_ids: &[String]) -> Result<Vec<SummaryInput>> {
+    use crate::openhuman::memory::tree::score::store::get_score;
+    use crate::openhuman::memory::tree::store::get_chunk;
+
+    let mut out: Vec<SummaryInput> = Vec::with_capacity(chunk_ids.len());
+    for id in chunk_ids {
+        let chunk = match get_chunk(config, id)? {
+            Some(c) => c,
+            None => {
+                log::warn!(
+                    "[source_tree::bucket_seal] hydrate_leaf_inputs: missing chunk {id} — skipping"
+                );
+                continue;
+            }
+        };
+        let score = get_score(config, id)?;
+        let (score_value, entities, topics) = match &score {
+            Some(row) => (row.total, Vec::new(), chunk.metadata.tags.clone()),
+            None => (0.0, Vec::new(), chunk.metadata.tags.clone()),
+        };
+        out.push(SummaryInput {
+            id: chunk.id.clone(),
+            content: chunk.content.clone(),
+            token_count: chunk.token_count,
+            entities,
+            topics,
+            time_range_start: chunk.metadata.time_range.0,
+            time_range_end: chunk.metadata.time_range.1,
+            score: score_value,
+        });
+    }
+    Ok(out)
+}
+
+fn hydrate_summary_inputs(config: &Config, summary_ids: &[String]) -> Result<Vec<SummaryInput>> {
+    let mut out: Vec<SummaryInput> = Vec::with_capacity(summary_ids.len());
+    for id in summary_ids {
+        let node = match store::get_summary(config, id)? {
+            Some(n) => n,
+            None => {
+                log::warn!(
+                    "[source_tree::bucket_seal] hydrate_summary_inputs: missing summary {id} — skipping"
+                );
+                continue;
+            }
+        };
+        out.push(SummaryInput {
+            id: node.id.clone(),
+            content: node.content.clone(),
+            token_count: node.token_count,
+            entities: node.entities.clone(),
+            topics: node.topics.clone(),
+            time_range_start: node.time_range_start,
+            time_range_end: node.time_range_end,
+            score: node.score,
+        });
+    }
+    Ok(out)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::openhuman::memory::tree::source_tree::registry::get_or_create_source_tree;
+    use crate::openhuman::memory::tree::source_tree::summariser::inert::InertSummariser;
+    use tempfile::TempDir;
+
+    fn test_config() -> (TempDir, Config) {
+        let tmp = TempDir::new().unwrap();
+        let mut cfg = Config::default();
+        cfg.workspace_dir = tmp.path().to_path_buf();
+        (tmp, cfg)
+    }
+
+    fn mk_leaf(id: &str, tokens: u32, ts_ms: i64) -> LeafRef {
+        use chrono::TimeZone;
+        LeafRef {
+            chunk_id: id.to_string(),
+            token_count: tokens,
+            timestamp: Utc.timestamp_millis_opt(ts_ms).single().unwrap(),
+            content: format!("content for {id}"),
+            entities: vec![],
+            topics: vec![],
+            score: 0.5,
+        }
+    }
+
+    #[tokio::test]
+    async fn append_below_budget_does_not_seal() {
+        let (_tmp, cfg) = test_config();
+        let tree = get_or_create_source_tree(&cfg, "slack:#eng").unwrap();
+        let summariser = InertSummariser::new();
+        // Chunks don't exist in DB — we're only exercising the buffer
+        // accounting, which doesn't require leaf rows until a seal fires.
+        let leaf = mk_leaf("leaf-1", 100, 1_700_000_000_000);
+        let sealed = append_leaf(&cfg, &tree, &leaf, &summariser).await.unwrap();
+        assert!(sealed.is_empty());
+
+        let buf = store::get_buffer(&cfg, &tree.id, 0).unwrap();
+        assert_eq!(buf.item_ids, vec!["leaf-1".to_string()]);
+        assert_eq!(buf.token_sum, 100);
+        assert_eq!(store::count_summaries(&cfg, &tree.id).unwrap(), 0);
+    }
+
+    #[tokio::test]
+    async fn crossing_budget_triggers_seal() {
+        use crate::openhuman::memory::tree::store::upsert_chunks;
+        use crate::openhuman::memory::tree::types::{
+            chunk_id, Chunk, Metadata, SourceKind, SourceRef,
+        };
+        use chrono::TimeZone;
+
+        let (_tmp, cfg) = test_config();
+        let tree = get_or_create_source_tree(&cfg, "slack:#eng").unwrap();
+        let summariser = InertSummariser::new();
+
+        // Persist two chunks that the hydrator can load during seal.
+        let ts = Utc.timestamp_millis_opt(1_700_000_000_000).unwrap();
+        let mk_chunk = |seq: u32, tokens: u32| Chunk {
+            id: chunk_id(SourceKind::Chat, "slack:#eng", seq),
+            content: format!("substantive chunk content {seq}"),
+            metadata: Metadata {
+                source_kind: SourceKind::Chat,
+                source_id: "slack:#eng".into(),
+                owner: "alice".into(),
+                timestamp: ts,
+                time_range: (ts, ts),
+                tags: vec![],
+                source_ref: Some(SourceRef::new("slack://x")),
+            },
+            token_count: tokens,
+            seq_in_source: seq,
+            created_at: ts,
+        };
+        let c1 = mk_chunk(0, 6_000);
+        let c2 = mk_chunk(1, 6_000);
+        upsert_chunks(&cfg, &[c1.clone(), c2.clone()]).unwrap();
+
+        // Two leaves whose combined token_sum (12k) exceeds the 10k budget.
+        let leaf1 = LeafRef {
+            chunk_id: c1.id.clone(),
+            token_count: 6_000,
+            timestamp: ts,
+            content: c1.content.clone(),
+            entities: vec![],
+            topics: vec![],
+            score: 0.5,
+        };
+        let leaf2 = LeafRef {
+            chunk_id: c2.id.clone(),
+            token_count: 6_000,
+            timestamp: ts,
+            content: c2.content.clone(),
+            entities: vec![],
+            topics: vec![],
+            score: 0.5,
+        };
+
+        let first = append_leaf(&cfg, &tree, &leaf1, &summariser).await.unwrap();
+        assert!(first.is_empty(), "first append below budget — no seal");
+
+        let second = append_leaf(&cfg, &tree, &leaf2, &summariser).await.unwrap();
+        assert_eq!(second.len(), 1, "second append crosses budget — one seal");
+
+        let summary_id = &second[0];
+        let summary = store::get_summary(&cfg, summary_id).unwrap().unwrap();
+        assert_eq!(summary.level, 1);
+        assert_eq!(summary.child_ids, vec![c1.id.clone(), c2.id.clone()]);
+        assert!(summary.token_count > 0);
+
+        // L0 buffer cleared, L1 buffer carries the new summary id.
+        let l0 = store::get_buffer(&cfg, &tree.id, 0).unwrap();
+        assert!(l0.is_empty());
+        let l1 = store::get_buffer(&cfg, &tree.id, 1).unwrap();
+        assert_eq!(l1.item_ids, vec![summary_id.clone()]);
+
+        // Tree metadata updated.
+        let t = store::get_tree(&cfg, &tree.id).unwrap().unwrap();
+        assert_eq!(t.max_level, 1);
+        assert_eq!(t.root_id.as_deref(), Some(summary_id.as_str()));
+        assert!(t.last_sealed_at.is_some());
+
+        // Leaf → parent backlink populated for both children.
+        use crate::openhuman::memory::tree::store::with_connection;
+        let parent: Option<String> = with_connection(&cfg, |conn| {
+            let p: Option<String> = conn
+                .query_row(
+                    "SELECT parent_summary_id FROM mem_tree_chunks WHERE id = ?1",
+                    rusqlite::params![c1.id],
+                    |r| r.get(0),
+                )
+                .unwrap();
+            Ok(p)
+        })
+        .unwrap();
+        assert_eq!(parent.as_deref(), Some(summary_id.as_str()));
+    }
+}

src/openhuman/memory/tree/source_tree/flush.rs

@@ -0,0 +1,192 @@
+//! Time-based buffer flush for source trees (#709).
+//!
+//! The bucket-seal path only fires when a buffer crosses `TOKEN_BUDGET`.
+//! Low-volume sources (e.g. an email account with two threads a week) can
+//! otherwise leave leaves parked in the L0 buffer indefinitely, which
+//! hurts recall. `flush_stale_buffers` force-seals any buffer whose
+//! `oldest_at` is older than `max_age`, regardless of token count.
+//!
+//! This is meant to run on a cadence (e.g. daily cron). Phase 3a ships
+//! the primitive; wiring into a scheduler is not required for merge.
+
+use anyhow::Result;
+use chrono::{DateTime, Duration, Utc};
+
+use crate::openhuman::config::Config;
+use crate::openhuman::memory::tree::source_tree::bucket_seal::cascade_all_from;
+use crate::openhuman::memory::tree::source_tree::store;
+use crate::openhuman::memory::tree::source_tree::summariser::Summariser;
+use crate::openhuman::memory::tree::source_tree::types::DEFAULT_FLUSH_AGE_SECS;
+
+/// Seal every buffer whose oldest item is older than `max_age`. Returns
+/// the number of individual seal calls (not trees) that fired. When the
+/// same tree has multiple stale levels they're each sealed in order.
+pub async fn flush_stale_buffers(
+    config: &Config,
+    max_age: Duration,
+    summariser: &dyn Summariser,
+) -> Result<usize> {
+    let now = Utc::now();
+    let cutoff = now - max_age;
+    let stale = store::list_stale_buffers(config, cutoff)?;
+    log::info!(
+        "[source_tree::flush] found {} stale buffers (max_age={:?})",
+        stale.len(),
+        max_age
+    );
+
+    let mut seals: usize = 0;
+    for buf in stale {
+        let tree = match store::get_tree(config, &buf.tree_id)? {
+            Some(t) => t,
+            None => {
+                log::warn!(
+                    "[source_tree::flush] orphan buffer tree_id={} level={}",
+                    buf.tree_id,
+                    buf.level
+                );
+                continue;
+            }
+        };
+        let sealed = cascade_all_from(config, &tree, buf.level, summariser, Some(now)).await?;
+        seals += sealed.len();
+    }
+    Ok(seals)
+}
+
+/// Convenience wrapper that uses [`DEFAULT_FLUSH_AGE_SECS`].
+pub async fn flush_stale_buffers_default(
+    config: &Config,
+    summariser: &dyn Summariser,
+) -> Result<usize> {
+    flush_stale_buffers(
+        config,
+        Duration::seconds(DEFAULT_FLUSH_AGE_SECS),
+        summariser,
+    )
+    .await
+}
+
+/// Helper exposed for callers that want a single explicit force-seal (e.g.
+/// "user disconnected this account, flush its buffer now").
+pub async fn force_flush_tree(
+    config: &Config,
+    tree_id: &str,
+    summariser: &dyn Summariser,
+    now: Option<DateTime<Utc>>,
+) -> Result<Vec<String>> {
+    let tree = store::get_tree(config, tree_id)?
+        .ok_or_else(|| anyhow::anyhow!("no tree with id {tree_id}"))?;
+    cascade_all_from(config, &tree, 0, summariser, now).await
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::openhuman::memory::tree::source_tree::bucket_seal::{append_leaf, LeafRef};
+    use crate::openhuman::memory::tree::source_tree::registry::get_or_create_source_tree;
+    use crate::openhuman::memory::tree::source_tree::summariser::inert::InertSummariser;
+    use crate::openhuman::memory::tree::store::upsert_chunks;
+    use crate::openhuman::memory::tree::types::{
+        chunk_id, Chunk, Metadata, SourceKind, SourceRef,
+    };
+    use tempfile::TempDir;
+
+    fn test_config() -> (TempDir, Config) {
+        let tmp = TempDir::new().unwrap();
+        let mut cfg = Config::default();
+        cfg.workspace_dir = tmp.path().to_path_buf();
+        (tmp, cfg)
+    }
+
+    #[tokio::test]
+    async fn flush_seals_old_buffer_even_under_budget() {
+        let (_tmp, cfg) = test_config();
+        let tree = get_or_create_source_tree(&cfg, "slack:#eng").unwrap();
+        let summariser = InertSummariser::new();
+
+        // Persist one chunk with an old timestamp (10 days ago).
+        let old_ts = Utc::now() - Duration::days(10);
+        let c = Chunk {
+            id: chunk_id(SourceKind::Chat, "slack:#eng", 0),
+            content: "old content that should get sealed".into(),
+            metadata: Metadata {
+                source_kind: SourceKind::Chat,
+                source_id: "slack:#eng".into(),
+                owner: "alice".into(),
+                timestamp: old_ts,
+                time_range: (old_ts, old_ts),
+                tags: vec![],
+                source_ref: Some(SourceRef::new("slack://x")),
+            },
+            token_count: 100,
+            seq_in_source: 0,
+            created_at: old_ts,
+        };
+        upsert_chunks(&cfg, &[c.clone()]).unwrap();
+
+        let leaf = LeafRef {
+            chunk_id: c.id.clone(),
+            token_count: 100, // way under the 10k budget
+            timestamp: old_ts,
+            content: c.content.clone(),
+            entities: vec![],
+            topics: vec![],
+            score: 0.5,
+        };
+        append_leaf(&cfg, &tree, &leaf, &summariser).await.unwrap();
+        assert_eq!(store::count_summaries(&cfg, &tree.id).unwrap(), 0);
+
+        let seals = flush_stale_buffers(&cfg, Duration::days(7), &summariser)
+            .await
+            .unwrap();
+        assert_eq!(seals, 1);
+        assert_eq!(store::count_summaries(&cfg, &tree.id).unwrap(), 1);
+
+        let l0 = store::get_buffer(&cfg, &tree.id, 0).unwrap();
+        assert!(l0.is_empty());
+    }
+
+    #[tokio::test]
+    async fn flush_noop_when_buffer_is_recent() {
+        let (_tmp, cfg) = test_config();
+        let tree = get_or_create_source_tree(&cfg, "slack:#eng").unwrap();
+        let summariser = InertSummariser::new();
+
+        // Persist a leaf stamped now so it's NOT stale.
+        let now = Utc::now();
+        let c = Chunk {
+            id: chunk_id(SourceKind::Chat, "slack:#eng", 0),
+            content: "fresh".into(),
+            metadata: Metadata {
+                source_kind: SourceKind::Chat,
+                source_id: "slack:#eng".into(),
+                owner: "alice".into(),
+                timestamp: now,
+                time_range: (now, now),
+                tags: vec![],
+                source_ref: Some(SourceRef::new("slack://x")),
+            },
+            token_count: 50,
+            seq_in_source: 0,
+            created_at: now,
+        };
+        upsert_chunks(&cfg, &[c.clone()]).unwrap();
+        let leaf = LeafRef {
+            chunk_id: c.id.clone(),
+            token_count: 50,
+            timestamp: now,
+            content: c.content.clone(),
+            entities: vec![],
+            topics: vec![],
+            score: 0.5,
+        };
+        append_leaf(&cfg, &tree, &leaf, &summariser).await.unwrap();
+
+        let seals = flush_stale_buffers(&cfg, Duration::days(7), &summariser)
+            .await
+            .unwrap();
+        assert_eq!(seals, 0);
+        assert_eq!(store::count_summaries(&cfg, &tree.id).unwrap(), 0);
+    }
+}

src/openhuman/memory/tree/source_tree/mod.rs

@@ -0,0 +1,28 @@
+//! Phase 3a — summary trees + bucket-seal mechanics (#709).
+//!
+//! A thin orchestration layer on top of Phase 1 chunks and Phase 2 scores
+//! that lifts individual leaves into a hierarchy of sealed summary nodes,
+//! one tree per ingest source. See `docs/MEMORY_ARCHITECTURE_LLD.md` for
+//! the full design. The module is isolated from the legacy
+//! `openhuman::memory` layer and only depends on sibling `tree::*` modules.
+//!
+//! Public surface at Phase 3a:
+//! - [`registry::get_or_create_source_tree`] — idempotent tree lookup
+//! - [`bucket_seal::append_leaf`] — push a chunk into its tree, cascade-seal on budget
+//! - [`flush::flush_stale_buffers`] — time-based seal of buffers that never cross budget
+//! - [`summariser::inert::InertSummariser`] — deterministic fallback summariser
+//!
+//! Phases 3b / 3c will add `global` and `topic` trees; both reuse the
+//! storage and cascade primitives defined here.
+
+pub mod bucket_seal;
+pub mod flush;
+pub mod registry;
+pub mod store;
+pub mod summariser;
+pub mod types;
+
+pub use bucket_seal::{append_leaf, LeafRef};
+pub use registry::get_or_create_source_tree;
+pub use summariser::{inert::InertSummariser, Summariser};
+pub use types::{Buffer, SummaryNode, Tree, TreeKind, TreeStatus, TOKEN_BUDGET};

src/openhuman/memory/tree/source_tree/registry.rs

@@ -0,0 +1,99 @@
+//! Tree registry — get-or-create for source trees (#709).
+//!
+//! The registry is the entry point for the ingest path to look up the
+//! tree for a given (kind, scope). Phase 3a only touches source trees;
+//! topic / global trees will reuse the same `(kind, scope)` convention
+//! in Phases 3b / 3c.
+
+use anyhow::Result;
+use chrono::Utc;
+use uuid::Uuid;
+
+use crate::openhuman::config::Config;
+use crate::openhuman::memory::tree::source_tree::store;
+use crate::openhuman::memory::tree::source_tree::types::{Tree, TreeKind, TreeStatus};
+
+/// Look up the source tree for `scope`, or create a new one.
+///
+/// Scope format convention (Phase 3a): use the ingested chunk's
+/// `metadata.source_id` verbatim, so re-ingesting the same Slack channel
+/// or Gmail account keeps appending to the same tree.
+pub fn get_or_create_source_tree(config: &Config, scope: &str) -> Result<Tree> {
+    if let Some(existing) = store::get_tree_by_scope(config, TreeKind::Source, scope)? {
+        log::debug!(
+            "[source_tree::registry] found tree id={} scope={}",
+            existing.id,
+            scope
+        );
+        return Ok(existing);
+    }
+
+    let tree = Tree {
+        id: new_tree_id(TreeKind::Source),
+        kind: TreeKind::Source,
+        scope: scope.to_string(),
+        root_id: None,
+        max_level: 0,
+        status: TreeStatus::Active,
+        created_at: Utc::now(),
+        last_sealed_at: None,
+    };
+    store::insert_tree(config, &tree)?;
+    log::info!(
+        "[source_tree::registry] created source tree id={} scope={}",
+        tree.id,
+        scope
+    );
+    Ok(tree)
+}
+
+fn new_tree_id(kind: TreeKind) -> String {
+    format!("{}:{}", kind.as_str(), Uuid::new_v4())
+}
+
+/// Public id generator for summary nodes — exported so `bucket_seal` can
+/// share the same format (kept separate for readability; both use UUID v4
+/// suffixes to keep ids short but unambiguous).
+pub fn new_summary_id(level: u32) -> String {
+    format!("summary:L{}:{}", level, Uuid::new_v4())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use tempfile::TempDir;
+
+    fn test_config() -> (TempDir, Config) {
+        let tmp = TempDir::new().unwrap();
+        let mut cfg = Config::default();
+        cfg.workspace_dir = tmp.path().to_path_buf();
+        (tmp, cfg)
+    }
+
+    #[test]
+    fn get_or_create_is_idempotent_on_scope() {
+        let (_tmp, cfg) = test_config();
+        let first = get_or_create_source_tree(&cfg, "slack:#eng").unwrap();
+        let second = get_or_create_source_tree(&cfg, "slack:#eng").unwrap();
+        assert_eq!(first.id, second.id);
+        assert_eq!(first.kind, TreeKind::Source);
+        assert_eq!(first.status, TreeStatus::Active);
+    }
+
+    #[test]
+    fn different_scopes_yield_different_trees() {
+        let (_tmp, cfg) = test_config();
+        let a = get_or_create_source_tree(&cfg, "slack:#eng").unwrap();
+        let b = get_or_create_source_tree(&cfg, "gmail:user@example.com").unwrap();
+        assert_ne!(a.id, b.id);
+        assert_ne!(a.scope, b.scope);
+    }
+
+    #[test]
+    fn tree_id_has_expected_prefix() {
+        let id = new_tree_id(TreeKind::Source);
+        assert!(id.starts_with("source:"));
+        let sum_id = new_summary_id(3);
+        assert!(sum_id.starts_with("summary:L3:"));
+    }
+}

src/openhuman/memory/tree/source_tree/store.rs

@@ -0,0 +1,610 @@
+//! SQLite-backed persistence for Phase 3a summary trees (#709).
+//!
+//! Three tables (schema lives in the sibling `tree::store::SCHEMA`):
+//! - `mem_tree_trees`      — one row per tree (kind, scope, root, max_level)
+//! - `mem_tree_summaries`  — one row per sealed summary node (immutable)
+//! - `mem_tree_buffers`    — one row per unsealed frontier `(tree_id, level)`
+//!
+//! All timestamps are stored as milliseconds since the Unix epoch so we
+//! share the epoch convention with `mem_tree_chunks`. Writes are serialised
+//! through the sibling `tree::store::with_connection` so we inherit its
+//! busy-timeout, WAL, and schema-init behaviour.
+
+use anyhow::{Context, Result};
+use chrono::{DateTime, TimeZone, Utc};
+use rusqlite::{params, Connection, OptionalExtension, Transaction};
+
+use crate::openhuman::config::Config;
+use crate::openhuman::memory::tree::source_tree::types::{
+    Buffer, SummaryNode, Tree, TreeKind, TreeStatus,
+};
+use crate::openhuman::memory::tree::store::with_connection;
+
+fn ms_to_utc(ms: i64) -> rusqlite::Result<DateTime<Utc>> {
+    Utc.timestamp_millis_opt(ms).single().ok_or_else(|| {
+        rusqlite::Error::FromSqlConversionFailure(
+            0,
+            rusqlite::types::Type::Integer,
+            format!("invalid timestamp ms {ms}").into(),
+        )
+    })
+}
+
+// ── Tree rows ───────────────────────────────────────────────────────────
+
+/// Insert a new tree row. Fails if `(kind, scope)` already exists; callers
+/// that want "get or create" semantics should go through the `registry`.
+pub fn insert_tree(config: &Config, tree: &Tree) -> Result<()> {
+    with_connection(config, |conn| insert_tree_conn(conn, tree))
+}
+
+pub(crate) fn insert_tree_conn(conn: &Connection, tree: &Tree) -> Result<()> {
+    conn.execute(
+        "INSERT INTO mem_tree_trees (
+            id, kind, scope, root_id, max_level, status,
+            created_at_ms, last_sealed_at_ms
+        ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)",
+        params![
+            tree.id,
+            tree.kind.as_str(),
+            tree.scope,
+            tree.root_id,
+            tree.max_level,
+            tree.status.as_str(),
+            tree.created_at.timestamp_millis(),
+            tree.last_sealed_at.map(|t| t.timestamp_millis()),
+        ],
+    )
+    .with_context(|| format!("Failed to insert tree id={}", tree.id))?;
+    Ok(())
+}
+
+/// Fetch a tree by `(kind, scope)`. Returns `None` if no such tree exists.
+pub fn get_tree_by_scope(
+    config: &Config,
+    kind: TreeKind,
+    scope: &str,
+) -> Result<Option<Tree>> {
+    with_connection(config, |conn| get_tree_by_scope_conn(conn, kind, scope))
+}
+
+pub(crate) fn get_tree_by_scope_conn(
+    conn: &Connection,
+    kind: TreeKind,
+    scope: &str,
+) -> Result<Option<Tree>> {
+    let mut stmt = conn.prepare(
+        "SELECT id, kind, scope, root_id, max_level, status,
+                created_at_ms, last_sealed_at_ms
+           FROM mem_tree_trees WHERE kind = ?1 AND scope = ?2",
+    )?;
+    let row = stmt
+        .query_row(params![kind.as_str(), scope], row_to_tree)
+        .optional()
+        .context("Failed to query tree by scope")?;
+    Ok(row)
+}
+
+/// Fetch a tree by primary key id.
+pub fn get_tree(config: &Config, id: &str) -> Result<Option<Tree>> {
+    with_connection(config, |conn| {
+        let mut stmt = conn.prepare(
+            "SELECT id, kind, scope, root_id, max_level, status,
+                    created_at_ms, last_sealed_at_ms
+               FROM mem_tree_trees WHERE id = ?1",
+        )?;
+        let row = stmt
+            .query_row(params![id], row_to_tree)
+            .optional()
+            .context("Failed to query tree by id")?;
+        Ok(row)
+    })
+}
+
+pub(crate) fn update_tree_after_seal_tx(
+    tx: &Transaction<'_>,
+    tree_id: &str,
+    root_id: &str,
+    max_level: u32,
+    sealed_at: DateTime<Utc>,
+) -> Result<()> {
+    tx.execute(
+        "UPDATE mem_tree_trees
+            SET root_id = ?1,
+                max_level = ?2,
+                last_sealed_at_ms = ?3
+          WHERE id = ?4",
+        params![
+            root_id,
+            max_level,
+            sealed_at.timestamp_millis(),
+            tree_id,
+        ],
+    )
+    .with_context(|| format!("Failed to update tree {tree_id} after seal"))?;
+    Ok(())
+}
+
+fn row_to_tree(row: &rusqlite::Row<'_>) -> rusqlite::Result<Tree> {
+    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())
+    })?;
+    Ok(Tree {
+        id,
+        kind,
+        scope,
+        root_id,
+        max_level: max_level.max(0) as u32,
+        status,
+        created_at: ms_to_utc(created_ms)?,
+        last_sealed_at: last_sealed_ms.map(ms_to_utc).transpose()?,
+    })
+}
+
+// ── Summary nodes ───────────────────────────────────────────────────────
+
+/// Insert a sealed summary. Immutable — the caller must generate a fresh
+/// id per seal. Idempotent on the primary key so retries of the same seal
+/// transaction don't double-insert.
+pub(crate) fn insert_summary_tx(tx: &Transaction<'_>, node: &SummaryNode) -> Result<()> {
+    tx.execute(
+        "INSERT OR IGNORE INTO mem_tree_summaries (
+            id, tree_id, tree_kind, level, parent_id,
+            child_ids_json, content, token_count,
+            entities_json, topics_json,
+            time_range_start_ms, time_range_end_ms,
+            score, sealed_at_ms, deleted
+        ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15)",
+        params![
+            node.id,
+            node.tree_id,
+            node.tree_kind.as_str(),
+            node.level,
+            node.parent_id,
+            serde_json::to_string(&node.child_ids)?,
+            node.content,
+            node.token_count,
+            serde_json::to_string(&node.entities)?,
+            serde_json::to_string(&node.topics)?,
+            node.time_range_start.timestamp_millis(),
+            node.time_range_end.timestamp_millis(),
+            node.score,
+            node.sealed_at.timestamp_millis(),
+            node.deleted as i64,
+        ],
+    )
+    .with_context(|| format!("Failed to insert summary id={}", node.id))?;
+    Ok(())
+}
+
+/// Fetch one summary by id. Soft-deleted rows are returned with
+/// `deleted = true` so callers can decide filtering policy.
+pub fn get_summary(config: &Config, id: &str) -> Result<Option<SummaryNode>> {
+    with_connection(config, |conn| {
+        let mut stmt = conn.prepare(
+            "SELECT id, tree_id, tree_kind, level, parent_id,
+                    child_ids_json, content, token_count,
+                    entities_json, topics_json,
+                    time_range_start_ms, time_range_end_ms,
+                    score, sealed_at_ms, deleted
+               FROM mem_tree_summaries WHERE id = ?1",
+        )?;
+        let row = stmt
+            .query_row(params![id], row_to_summary)
+            .optional()
+            .context("Failed to query summary by id")?;
+        Ok(row)
+    })
+}
+
+/// List sealed summaries for a tree at a given level, ordered by
+/// `sealed_at` ascending. Skips tombstoned rows.
+pub fn list_summaries_at_level(
+    config: &Config,
+    tree_id: &str,
+    level: u32,
+) -> Result<Vec<SummaryNode>> {
+    with_connection(config, |conn| {
+        let mut stmt = conn.prepare(
+            "SELECT id, tree_id, tree_kind, level, parent_id,
+                    child_ids_json, content, token_count,
+                    entities_json, topics_json,
+                    time_range_start_ms, time_range_end_ms,
+                    score, sealed_at_ms, deleted
+               FROM mem_tree_summaries
+              WHERE tree_id = ?1 AND level = ?2 AND deleted = 0
+              ORDER BY sealed_at_ms ASC",
+        )?;
+        let rows = stmt
+            .query_map(params![tree_id, level], row_to_summary)?
+            .collect::<rusqlite::Result<Vec<_>>>()
+            .context("Failed to collect summaries")?;
+        Ok(rows)
+    })
+}
+
+/// Count summaries in a tree (diagnostic helper).
+pub fn count_summaries(config: &Config, tree_id: &str) -> Result<u64> {
+    with_connection(config, |conn| {
+        let n: i64 = conn
+            .query_row(
+                "SELECT COUNT(*) FROM mem_tree_summaries
+                  WHERE tree_id = ?1 AND deleted = 0",
+                params![tree_id],
+                |r| r.get(0),
+            )
+            .context("count summaries query")?;
+        Ok(n.max(0) as u64)
+    })
+}
+
+fn row_to_summary(row: &rusqlite::Row<'_>) -> rusqlite::Result<SummaryNode> {
+    let id: String = row.get(0)?;
+    let tree_id: String = row.get(1)?;
+    let tree_kind_s: String = row.get(2)?;
+    let level: i64 = row.get(3)?;
+    let parent_id: Option<String> = row.get(4)?;
+    let child_ids_json: String = row.get(5)?;
+    let content: String = row.get(6)?;
+    let token_count: i64 = row.get(7)?;
+    let entities_json: String = row.get(8)?;
+    let topics_json: String = row.get(9)?;
+    let trs_ms: i64 = row.get(10)?;
+    let tre_ms: i64 = row.get(11)?;
+    let score: f64 = row.get(12)?;
+    let sealed_ms: i64 = row.get(13)?;
+    let deleted: i64 = row.get(14)?;
+
+    let tree_kind = TreeKind::parse(&tree_kind_s).map_err(|e| {
+        rusqlite::Error::FromSqlConversionFailure(2, rusqlite::types::Type::Text, e.into())
+    })?;
+    let child_ids: Vec<String> = serde_json::from_str(&child_ids_json).map_err(|e| {
+        rusqlite::Error::FromSqlConversionFailure(5, rusqlite::types::Type::Text, Box::new(e))
+    })?;
+    let entities: Vec<String> = serde_json::from_str(&entities_json).map_err(|e| {
+        rusqlite::Error::FromSqlConversionFailure(8, rusqlite::types::Type::Text, Box::new(e))
+    })?;
+    let topics: Vec<String> = serde_json::from_str(&topics_json).map_err(|e| {
+        rusqlite::Error::FromSqlConversionFailure(9, rusqlite::types::Type::Text, Box::new(e))
+    })?;
+
+    Ok(SummaryNode {
+        id,
+        tree_id,
+        tree_kind,
+        level: level.max(0) as u32,
+        parent_id,
+        child_ids,
+        content,
+        token_count: token_count.max(0) as u32,
+        entities,
+        topics,
+        time_range_start: ms_to_utc(trs_ms)?,
+        time_range_end: ms_to_utc(tre_ms)?,
+        score: score as f32,
+        sealed_at: ms_to_utc(sealed_ms)?,
+        deleted: deleted != 0,
+    })
+}
+
+// ── Buffers ─────────────────────────────────────────────────────────────
+
+/// Read the current buffer at `(tree_id, level)` or return an empty one.
+pub fn get_buffer(config: &Config, tree_id: &str, level: u32) -> Result<Buffer> {
+    with_connection(config, |conn| get_buffer_conn(conn, tree_id, level))
+}
+
+pub(crate) fn get_buffer_conn(
+    conn: &Connection,
+    tree_id: &str,
+    level: u32,
+) -> Result<Buffer> {
+    let mut stmt = conn.prepare(
+        "SELECT tree_id, level, item_ids_json, token_sum, oldest_at_ms
+           FROM mem_tree_buffers WHERE tree_id = ?1 AND level = ?2",
+    )?;
+    let row = stmt
+        .query_row(params![tree_id, level], row_to_buffer)
+        .optional()
+        .context("Failed to query buffer")?;
+    Ok(row.unwrap_or_else(|| Buffer::empty(tree_id, level)))
+}
+
+/// Upsert a buffer row.
+pub(crate) fn upsert_buffer_tx(tx: &Transaction<'_>, buf: &Buffer) -> Result<()> {
+    let now_ms = Utc::now().timestamp_millis();
+    tx.execute(
+        "INSERT INTO mem_tree_buffers (
+            tree_id, level, item_ids_json, token_sum, oldest_at_ms, updated_at_ms
+        ) VALUES (?1, ?2, ?3, ?4, ?5, ?6)
+        ON CONFLICT(tree_id, level) DO UPDATE SET
+            item_ids_json = excluded.item_ids_json,
+            token_sum = excluded.token_sum,
+            oldest_at_ms = excluded.oldest_at_ms,
+            updated_at_ms = excluded.updated_at_ms",
+        params![
+            buf.tree_id,
+            buf.level,
+            serde_json::to_string(&buf.item_ids)?,
+            buf.token_sum,
+            buf.oldest_at.map(|t| t.timestamp_millis()),
+            now_ms,
+        ],
+    )
+    .with_context(|| {
+        format!(
+            "Failed to upsert buffer tree_id={} level={}",
+            buf.tree_id, buf.level
+        )
+    })?;
+    Ok(())
+}
+
+/// Reset a buffer at `(tree_id, level)` to empty. Used at seal time: the
+/// items move into a summary row and the buffer is cleared in the same tx.
+pub(crate) fn clear_buffer_tx(tx: &Transaction<'_>, tree_id: &str, level: u32) -> Result<()> {
+    let empty = Buffer::empty(tree_id, level);
+    upsert_buffer_tx(tx, &empty)
+}
+
+/// List all non-empty buffers ordered by `oldest_at_ms ASC`. Used by the
+/// time-based flush pass.
+pub fn list_stale_buffers(
+    config: &Config,
+    older_than: DateTime<Utc>,
+) -> Result<Vec<Buffer>> {
+    with_connection(config, |conn| {
+        let mut stmt = conn.prepare(
+            "SELECT tree_id, level, item_ids_json, token_sum, oldest_at_ms
+               FROM mem_tree_buffers
+              WHERE oldest_at_ms IS NOT NULL
+                AND oldest_at_ms <= ?1
+              ORDER BY oldest_at_ms ASC",
+        )?;
+        let rows = stmt
+            .query_map(params![older_than.timestamp_millis()], row_to_buffer)?
+            .collect::<rusqlite::Result<Vec<_>>>()
+            .context("Failed to collect stale buffers")?;
+        Ok(rows)
+    })
+}
+
+fn row_to_buffer(row: &rusqlite::Row<'_>) -> rusqlite::Result<Buffer> {
+    let tree_id: String = row.get(0)?;
+    let level: i64 = row.get(1)?;
+    let item_ids_json: String = row.get(2)?;
+    let token_sum: i64 = row.get(3)?;
+    let oldest_ms: Option<i64> = row.get(4)?;
+
+    let item_ids: Vec<String> = serde_json::from_str(&item_ids_json).map_err(|e| {
+        rusqlite::Error::FromSqlConversionFailure(2, rusqlite::types::Type::Text, Box::new(e))
+    })?;
+    let oldest_at = oldest_ms.map(ms_to_utc).transpose()?;
+    Ok(Buffer {
+        tree_id,
+        level: level.max(0) as u32,
+        item_ids,
+        token_sum,
+        oldest_at,
+    })
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use tempfile::TempDir;
+
+    fn test_config() -> (TempDir, Config) {
+        let tmp = TempDir::new().unwrap();
+        let mut cfg = Config::default();
+        cfg.workspace_dir = tmp.path().to_path_buf();
+        (tmp, cfg)
+    }
+
+    fn sample_tree(id: &str, scope: &str) -> Tree {
+        Tree {
+            id: id.to_string(),
+            kind: TreeKind::Source,
+            scope: scope.to_string(),
+            root_id: None,
+            max_level: 0,
+            status: TreeStatus::Active,
+            created_at: Utc.timestamp_millis_opt(1_700_000_000_000).unwrap(),
+            last_sealed_at: None,
+        }
+    }
+
+    fn sample_summary(id: &str, tree_id: &str, level: u32) -> SummaryNode {
+        let ts = Utc.timestamp_millis_opt(1_700_000_000_000).unwrap();
+        SummaryNode {
+            id: id.to_string(),
+            tree_id: tree_id.to_string(),
+            tree_kind: TreeKind::Source,
+            level,
+            parent_id: None,
+            child_ids: vec!["leaf-a".into(), "leaf-b".into()],
+            content: "seal content".into(),
+            token_count: 100,
+            entities: vec!["entity:alice".into()],
+            topics: vec!["#launch".into()],
+            time_range_start: ts,
+            time_range_end: ts,
+            score: 0.75,
+            sealed_at: ts,
+            deleted: false,
+        }
+    }
+
+    #[test]
+    fn tree_round_trip() {
+        let (_tmp, cfg) = test_config();
+        let t = sample_tree("tree-1", "slack:#eng");
+        insert_tree(&cfg, &t).unwrap();
+        let got = get_tree(&cfg, "tree-1").unwrap().unwrap();
+        assert_eq!(got, t);
+        let by_scope = get_tree_by_scope(&cfg, TreeKind::Source, "slack:#eng")
+            .unwrap()
+            .unwrap();
+        assert_eq!(by_scope.id, "tree-1");
+    }
+
+    #[test]
+    fn duplicate_scope_fails() {
+        let (_tmp, cfg) = test_config();
+        insert_tree(&cfg, &sample_tree("t1", "slack:#eng")).unwrap();
+        let dup = sample_tree("t2", "slack:#eng");
+        assert!(insert_tree(&cfg, &dup).is_err());
+    }
+
+    #[test]
+    fn summary_insert_and_fetch() {
+        let (_tmp, cfg) = test_config();
+        insert_tree(&cfg, &sample_tree("tree-1", "slack:#eng")).unwrap();
+        let node = sample_summary("sum-1", "tree-1", 1);
+        with_connection(&cfg, |conn| {
+            let tx = conn.unchecked_transaction()?;
+            insert_summary_tx(&tx, &node)?;
+            tx.commit()?;
+            Ok(())
+        })
+        .unwrap();
+        let got = get_summary(&cfg, "sum-1").unwrap().unwrap();
+        assert_eq!(got, node);
+        let at_level = list_summaries_at_level(&cfg, "tree-1", 1).unwrap();
+        assert_eq!(at_level.len(), 1);
+        assert_eq!(count_summaries(&cfg, "tree-1").unwrap(), 1);
+    }
+
+    #[test]
+    fn summary_insert_is_idempotent_on_id() {
+        let (_tmp, cfg) = test_config();
+        insert_tree(&cfg, &sample_tree("tree-1", "slack:#eng")).unwrap();
+        let node = sample_summary("sum-1", "tree-1", 1);
+        with_connection(&cfg, |conn| {
+            let tx = conn.unchecked_transaction()?;
+            insert_summary_tx(&tx, &node)?;
+            insert_summary_tx(&tx, &node)?;
+            tx.commit()?;
+            Ok(())
+        })
+        .unwrap();
+        assert_eq!(count_summaries(&cfg, "tree-1").unwrap(), 1);
+    }
+
+    #[test]
+    fn buffer_upsert_and_clear() {
+        let (_tmp, cfg) = test_config();
+        insert_tree(&cfg, &sample_tree("tree-1", "slack:#eng")).unwrap();
+        let ts = Utc.timestamp_millis_opt(1_700_000_000_000).unwrap();
+        let buf = Buffer {
+            tree_id: "tree-1".into(),
+            level: 0,
+            item_ids: vec!["leaf-a".into(), "leaf-b".into()],
+            token_sum: 500,
+            oldest_at: Some(ts),
+        };
+        with_connection(&cfg, |conn| {
+            let tx = conn.unchecked_transaction()?;
+            upsert_buffer_tx(&tx, &buf)?;
+            tx.commit()?;
+            Ok(())
+        })
+        .unwrap();
+        let got = get_buffer(&cfg, "tree-1", 0).unwrap();
+        assert_eq!(got, buf);
+
+        with_connection(&cfg, |conn| {
+            let tx = conn.unchecked_transaction()?;
+            clear_buffer_tx(&tx, "tree-1", 0)?;
+            tx.commit()?;
+            Ok(())
+        })
+        .unwrap();
+        let cleared = get_buffer(&cfg, "tree-1", 0).unwrap();
+        assert!(cleared.is_empty());
+        assert_eq!(cleared.token_sum, 0);
+        assert!(cleared.oldest_at.is_none());
+    }
+
+    #[test]
+    fn get_buffer_returns_empty_when_missing() {
+        let (_tmp, cfg) = test_config();
+        insert_tree(&cfg, &sample_tree("tree-1", "slack:#eng")).unwrap();
+        let got = get_buffer(&cfg, "tree-1", 0).unwrap();
+        assert!(got.is_empty());
+        assert_eq!(got.tree_id, "tree-1");
+    }
+
+    #[test]
+    fn update_tree_after_seal_persists() {
+        let (_tmp, cfg) = test_config();
+        insert_tree(&cfg, &sample_tree("tree-1", "slack:#eng")).unwrap();
+        let sealed_at = Utc.timestamp_millis_opt(1_700_000_123_000).unwrap();
+        with_connection(&cfg, |conn| {
+            let tx = conn.unchecked_transaction()?;
+            update_tree_after_seal_tx(&tx, "tree-1", "sum-1", 1, sealed_at)?;
+            tx.commit()?;
+            Ok(())
+        })
+        .unwrap();
+        let got = get_tree(&cfg, "tree-1").unwrap().unwrap();
+        assert_eq!(got.root_id.as_deref(), Some("sum-1"));
+        assert_eq!(got.max_level, 1);
+        assert_eq!(got.last_sealed_at, Some(sealed_at));
+    }
+
+    #[test]
+    fn list_stale_buffers_orders_by_age() {
+        let (_tmp, cfg) = test_config();
+        insert_tree(&cfg, &sample_tree("tree-1", "slack:#eng")).unwrap();
+        let t0 = Utc.timestamp_millis_opt(1_700_000_000_000).unwrap();
+        let t1 = Utc.timestamp_millis_opt(1_700_000_010_000).unwrap();
+        let t2 = Utc.timestamp_millis_opt(1_700_000_020_000).unwrap();
+        with_connection(&cfg, |conn| {
+            let tx = conn.unchecked_transaction()?;
+            upsert_buffer_tx(
+                &tx,
+                &Buffer {
+                    tree_id: "tree-1".into(),
+                    level: 0,
+                    item_ids: vec!["a".into()],
+                    token_sum: 10,
+                    oldest_at: Some(t0),
+                },
+            )?;
+            upsert_buffer_tx(
+                &tx,
+                &Buffer {
+                    tree_id: "tree-1".into(),
+                    level: 1,
+                    item_ids: vec!["b".into()],
+                    token_sum: 20,
+                    oldest_at: Some(t1),
+                },
+            )?;
+            tx.commit()?;
+            Ok(())
+        })
+        .unwrap();
+        let stale = list_stale_buffers(&cfg, t2).unwrap();
+        assert_eq!(stale.len(), 2);
+        assert_eq!(stale[0].level, 0);
+        assert_eq!(stale[1].level, 1);
+        // Filter out the first: only level-1 should come back.
+        let only_later = list_stale_buffers(&cfg, t0).unwrap();
+        assert_eq!(only_later.len(), 1);
+        assert_eq!(only_later[0].level, 0);
+    }
+}

src/openhuman/memory/tree/source_tree/summariser/inert.rs

@@ -0,0 +1,190 @@
+//! Deterministic fallback summariser (#709).
+//!
+//! `InertSummariser` concatenates each input's content, separated by a
+//! blank line, and hard-truncates to `ctx.token_budget`. It also unions
+//! the entity and topic sets (dedup while preserving first-seen order).
+//! The goal is not fidelity — it's a stable, dependency-free baseline so
+//! tree mechanics (sealing, cascade, roots) can be tested without an LLM.
+
+use anyhow::Result;
+use async_trait::async_trait;
+
+use crate::openhuman::memory::tree::source_tree::summariser::{
+    SummaryContext, SummaryInput, SummaryOutput, Summariser,
+};
+use crate::openhuman::memory::tree::types::approx_token_count;
+
+/// Default prefix applied to each contribution in the joined body. Keeps
+/// provenance visible to a human reading the raw summary.
+const PROVENANCE_PREFIX: &str = "— ";
+
+pub struct InertSummariser;
+
+impl InertSummariser {
+    pub fn new() -> Self {
+        Self
+    }
+}
+
+impl Default for InertSummariser {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+#[async_trait]
+impl Summariser for InertSummariser {
+    async fn summarise(
+        &self,
+        inputs: &[SummaryInput],
+        ctx: &SummaryContext<'_>,
+    ) -> Result<SummaryOutput> {
+        let mut parts: Vec<String> = Vec::with_capacity(inputs.len());
+        for inp in inputs {
+            let trimmed = inp.content.trim();
+            if trimmed.is_empty() {
+                continue;
+            }
+            parts.push(format!("{}{}", PROVENANCE_PREFIX, trimmed));
+        }
+        let joined = parts.join("\n\n");
+
+        let (content, token_count) = truncate_to_budget(&joined, ctx.token_budget);
+
+        let entities = union_preserve_order(inputs.iter().map(|i| i.entities.as_slice()));
+        let topics = union_preserve_order(inputs.iter().map(|i| i.topics.as_slice()));
+
+        log::debug!(
+            "[source_tree::summariser::inert] sealed tree_id={} level={} inputs={} tokens={}",
+            ctx.tree_id,
+            ctx.target_level,
+            inputs.len(),
+            token_count
+        );
+
+        Ok(SummaryOutput {
+            content,
+            token_count,
+            entities,
+            topics,
+        })
+    }
+}
+
+/// Truncate `text` to fit within `budget` approximate tokens. Returns the
+/// (possibly truncated) body and its recomputed token count. Truncation is
+/// done on character boundaries — `approx_token_count` assumes ~4 chars
+/// per token so we clamp character length to `budget * 4`.
+fn truncate_to_budget(text: &str, budget: u32) -> (String, u32) {
+    let initial = approx_token_count(text);
+    if initial <= budget {
+        return (text.to_string(), initial);
+    }
+    // Character ceiling derived from the same ~4 chars/token heuristic.
+    let char_ceiling = (budget as usize).saturating_mul(4);
+    let truncated: String = text.chars().take(char_ceiling).collect();
+    let tokens = approx_token_count(&truncated);
+    (truncated, tokens)
+}
+
+fn union_preserve_order<'a, I>(iter: I) -> Vec<String>
+where
+    I: IntoIterator<Item = &'a [String]>,
+{
+    let mut out: Vec<String> = Vec::new();
+    for group in iter {
+        for item in group {
+            if !out.iter().any(|existing| existing == item) {
+                out.push(item.clone());
+            }
+        }
+    }
+    out
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::openhuman::memory::tree::source_tree::types::TreeKind;
+    use chrono::Utc;
+
+    fn sample_input(id: &str, content: &str, entities: &[&str]) -> SummaryInput {
+        let ts = Utc::now();
+        SummaryInput {
+            id: id.to_string(),
+            content: content.to_string(),
+            token_count: approx_token_count(content),
+            entities: entities.iter().map(|s| s.to_string()).collect(),
+            topics: Vec::new(),
+            time_range_start: ts,
+            time_range_end: ts,
+            score: 0.5,
+        }
+    }
+
+    fn test_ctx() -> SummaryContext<'static> {
+        SummaryContext {
+            tree_id: "tree-1",
+            tree_kind: TreeKind::Source,
+            target_level: 1,
+            token_budget: 10_000,
+        }
+    }
+
+    #[tokio::test]
+    async fn concats_inputs_with_provenance_prefix() {
+        let s = InertSummariser::default();
+        let inputs = vec![
+            sample_input("a", "hello world", &[]),
+            sample_input("b", "second contribution", &[]),
+        ];
+        let out = s.summarise(&inputs, &test_ctx()).await.unwrap();
+        assert!(out.content.contains(PROVENANCE_PREFIX));
+        assert!(out.content.contains("hello world"));
+        assert!(out.content.contains("second contribution"));
+        assert_eq!(out.token_count, approx_token_count(&out.content));
+    }
+
+    #[tokio::test]
+    async fn unions_entities_preserving_order_and_dedupe() {
+        let s = InertSummariser::default();
+        let inputs = vec![
+            sample_input("a", "x", &["entity:alice", "entity:bob"]),
+            sample_input("b", "y", &["entity:bob", "entity:carol"]),
+        ];
+        let out = s.summarise(&inputs, &test_ctx()).await.unwrap();
+        assert_eq!(
+            out.entities,
+            vec![
+                "entity:alice".to_string(),
+                "entity:bob".to_string(),
+                "entity:carol".to_string(),
+            ]
+        );
+    }
+
+    #[tokio::test]
+    async fn truncates_when_over_budget() {
+        let s = InertSummariser::default();
+        let long_text = "a".repeat(100);
+        let inputs = vec![sample_input("a", &long_text, &[])];
+        let mut ctx = test_ctx();
+        ctx.token_budget = 5; // way under — should truncate hard
+        let out = s.summarise(&inputs, &ctx).await.unwrap();
+        assert!(out.token_count <= ctx.token_budget + 1);
+        assert!(out.content.len() < long_text.len() + PROVENANCE_PREFIX.len());
+    }
+
+    #[tokio::test]
+    async fn skips_empty_contributions() {
+        let s = InertSummariser::default();
+        let inputs = vec![
+            sample_input("a", "   ", &[]),
+            sample_input("b", "kept", &[]),
+        ];
+        let out = s.summarise(&inputs, &test_ctx()).await.unwrap();
+        assert!(out.content.contains("kept"));
+        // exactly one provenance prefix should appear
+        assert_eq!(out.content.matches(PROVENANCE_PREFIX).count(), 1);
+    }
+}

src/openhuman/memory/tree/source_tree/summariser/mod.rs

@@ -0,0 +1,62 @@
+//! Summariser trait + fallback (#709).
+//!
+//! A summariser folds N buffered items into one sealed summary. Phase 3a
+//! ships an `InertSummariser` that concatenates the contributions and
+//! truncates to the token budget — enough to make the tree mechanics
+//! observable end-to-end without requiring an LLM. Real summarisation
+//! (Ollama, etc.) can slot in by implementing the trait.
+
+use anyhow::Result;
+use async_trait::async_trait;
+use chrono::{DateTime, Utc};
+
+use crate::openhuman::memory::tree::source_tree::types::TreeKind;
+
+pub mod inert;
+
+/// One contribution being folded — either a raw leaf (chunk) at L0→L1, or
+/// a lower-level summary at L_n→L_{n+1}.
+#[derive(Clone, Debug)]
+pub struct SummaryInput {
+    /// Primary key of the contribution (chunk id or summary id).
+    pub id: String,
+    pub content: String,
+    pub token_count: u32,
+    pub entities: Vec<String>,
+    pub topics: Vec<String>,
+    pub time_range_start: DateTime<Utc>,
+    pub time_range_end: DateTime<Utc>,
+    /// Score signal from scoring (for leaves) or parent seal (for summaries).
+    pub score: f32,
+}
+
+/// Opaque context passed to the summariser — lets implementations log /
+/// identify which tree is being sealed without threading config globally.
+#[derive(Clone, Debug)]
+pub struct SummaryContext<'a> {
+    pub tree_id: &'a str,
+    pub tree_kind: TreeKind,
+    pub target_level: u32,
+    pub token_budget: u32,
+}
+
+/// Output of a summariser invocation.
+#[derive(Clone, Debug)]
+pub struct SummaryOutput {
+    pub content: String,
+    pub token_count: u32,
+    pub entities: Vec<String>,
+    pub topics: Vec<String>,
+}
+
+#[async_trait]
+pub trait Summariser: Send + Sync {
+    /// Fold the inputs into a single summary. `ctx.token_budget` is an
+    /// upper bound on the produced `token_count`; implementations SHOULD
+    /// stay well under it so parents have room to include this summary.
+    async fn summarise(
+        &self,
+        inputs: &[SummaryInput],
+        ctx: &SummaryContext<'_>,
+    ) -> Result<SummaryOutput>;
+}

src/openhuman/memory/tree/source_tree/types.rs

@@ -0,0 +1,215 @@
+//! Core types for Phase 3a — summary trees, per-source bucket-seal (#709).
+//!
+//! These types sit on top of Phase 1's chunk leaves. A [`Tree`] groups leaves
+//! under one scope (e.g. one chat channel, one email account). When a
+//! [`Buffer`] at some level accumulates enough tokens, its contents seal
+//! into a [`SummaryNode`] at level+1 and the buffer clears. Summary nodes
+//! are immutable once emitted — updates to children use the Phase 1/2
+//! tombstone pattern, never rewrite parents.
+
+use chrono::{DateTime, Utc};
+use serde::{Deserialize, Serialize};
+
+/// What kind of tree this is. Source trees live per ingest source; topic
+/// and global trees are introduced in Phase 3b/3c and share the same
+/// schema.
+#[derive(Clone, Copy, Debug, Eq, PartialEq, Hash, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+#[non_exhaustive]
+pub enum TreeKind {
+    /// One tree per ingest source (e.g. `chat:slack:#eng`, `email:gmail:user`).
+    Source,
+    /// Reserved for Phase 3c — per-entity/topic tree.
+    Topic,
+    /// Reserved for Phase 3b — cross-source daily digest tree.
+    Global,
+}
+
+impl TreeKind {
+    pub fn as_str(self) -> &'static str {
+        match self {
+            Self::Source => "source",
+            Self::Topic => "topic",
+            Self::Global => "global",
+        }
+    }
+
+    pub fn parse(s: &str) -> Result<Self, String> {
+        match s {
+            "source" => Ok(Self::Source),
+            "topic" => Ok(Self::Topic),
+            "global" => Ok(Self::Global),
+            other => Err(format!("unknown tree kind: {other}")),
+        }
+    }
+}
+
+/// Activity state of a tree. Archived trees stay queryable but don't accept
+/// new leaves — used by Phase 3c when a topic tree's entity goes cold.
+#[derive(Clone, Copy, Debug, Eq, PartialEq, Hash, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum TreeStatus {
+    Active,
+    Archived,
+}
+
+impl TreeStatus {
+    pub fn as_str(self) -> &'static str {
+        match self {
+            Self::Active => "active",
+            Self::Archived => "archived",
+        }
+    }
+
+    pub fn parse(s: &str) -> Result<Self, String> {
+        match s {
+            "active" => Ok(Self::Active),
+            "archived" => Ok(Self::Archived),
+            other => Err(format!("unknown tree status: {other}")),
+        }
+    }
+}
+
+/// One summary-tree instance.
+///
+/// `root_id` is `None` until the first seal emits an L1 node. `max_level`
+/// tracks the highest level that has ever sealed; `root_id` points at the
+/// current top node at that level (changes on root-split).
+#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
+pub struct Tree {
+    pub id: String,
+    pub kind: TreeKind,
+    /// Logical identifier for what the tree covers. Format conventions:
+    /// - Source: `<source_kind>:<provider>:<source_id>` or the chunk's
+    ///   `source_id` directly (Phase 3a uses the chunk source_id verbatim)
+    /// - Topic: canonical entity id
+    /// - Global: the literal string `"global"`
+    pub scope: String,
+    pub root_id: Option<String>,
+    pub max_level: u32,
+    pub status: TreeStatus,
+    pub created_at: DateTime<Utc>,
+    pub last_sealed_at: Option<DateTime<Utc>>,
+}
+
+/// A sealed summary node — one level above raw leaves.
+///
+/// `child_ids` points at the concrete children that were in the buffer when
+/// this node sealed. For L1 nodes those are leaf `chunk.id`s; for L2+ they
+/// are lower-level summary ids. Relation is fixed at seal time — never
+/// modified afterwards.
+#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
+pub struct SummaryNode {
+    pub id: String,
+    pub tree_id: String,
+    pub tree_kind: TreeKind,
+    /// 1 for summaries over raw leaves, 2 over L1 summaries, and so on.
+    pub level: u32,
+    pub parent_id: Option<String>,
+    pub child_ids: Vec<String>,
+    /// Summariser output. Typical target: 800–1500 tokens.
+    pub content: String,
+    pub token_count: u32,
+    /// Curated subset of children's entity canonical-ids.
+    pub entities: Vec<String>,
+    /// Curated topic labels (hashtag-like short phrases).
+    pub topics: Vec<String>,
+    pub time_range_start: DateTime<Utc>,
+    pub time_range_end: DateTime<Utc>,
+    /// Max of children's scores at seal time — cheap heuristic, preserved
+    /// for reranking in Phase 4.
+    pub score: f32,
+    pub sealed_at: DateTime<Utc>,
+    /// Tombstone flag — stays `false` in Phase 3a since summaries are
+    /// immutable. Reserved for future cleanup passes (e.g. archive cascade).
+    pub deleted: bool,
+}
+
+/// Unsealed frontier at a given `(tree_id, level)`. One row per level per
+/// tree. `oldest_at` is `None` when the buffer is empty; used by the
+/// time-based flush trigger.
+#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
+pub struct Buffer {
+    pub tree_id: String,
+    pub level: u32,
+    pub item_ids: Vec<String>,
+    pub token_sum: i64,
+    pub oldest_at: Option<DateTime<Utc>>,
+}
+
+impl Buffer {
+    /// Empty buffer at the given key.
+    pub fn empty(tree_id: &str, level: u32) -> Self {
+        Self {
+            tree_id: tree_id.to_string(),
+            level,
+            item_ids: Vec::new(),
+            token_sum: 0,
+            oldest_at: None,
+        }
+    }
+
+    pub fn is_empty(&self) -> bool {
+        self.item_ids.is_empty()
+    }
+
+    /// Whether the buffer's oldest item is older than `max_age`. Returns
+    /// `false` for an empty buffer.
+    pub fn is_stale(&self, now: DateTime<Utc>, max_age: chrono::Duration) -> bool {
+        match self.oldest_at {
+            Some(ts) => now.signed_duration_since(ts) > max_age,
+            None => false,
+        }
+    }
+}
+
+/// Token ceiling for one summariser invocation — aligned with the Phase 1
+/// chunker ceiling so a single leaf never busts a seal on its own.
+pub const TOKEN_BUDGET: u32 = 10_000;
+
+/// Default age at which a non-empty buffer is force-sealed even under the
+/// token budget. Keeps recent activity from stalling waiting for more
+/// leaves that may never arrive.
+pub const DEFAULT_FLUSH_AGE_SECS: i64 = 7 * 24 * 60 * 60;
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn tree_kind_round_trip() {
+        for k in [TreeKind::Source, TreeKind::Topic, TreeKind::Global] {
+            assert_eq!(TreeKind::parse(k.as_str()).unwrap(), k);
+        }
+        assert!(TreeKind::parse("bogus").is_err());
+    }
+
+    #[test]
+    fn tree_status_round_trip() {
+        for s in [TreeStatus::Active, TreeStatus::Archived] {
+            assert_eq!(TreeStatus::parse(s.as_str()).unwrap(), s);
+        }
+        assert!(TreeStatus::parse("live").is_err());
+    }
+
+    #[test]
+    fn empty_buffer_is_not_stale() {
+        let b = Buffer::empty("t1", 0);
+        assert!(b.is_empty());
+        assert!(!b.is_stale(Utc::now(), chrono::Duration::zero()));
+    }
+
+    #[test]
+    fn stale_buffer_detected() {
+        let past = Utc::now() - chrono::Duration::hours(10);
+        let b = Buffer {
+            tree_id: "t1".into(),
+            level: 0,
+            item_ids: vec!["leaf-1".into()],
+            token_sum: 100,
+            oldest_at: Some(past),
+        };
+        assert!(b.is_stale(Utc::now(), chrono::Duration::hours(1)));
+        assert!(!b.is_stale(Utc::now(), chrono::Duration::hours(20)));
+    }
+}

src/openhuman/memory/tree/store.rs

@@ -88,6 +88,71 @@ CREATE INDEX IF NOT EXISTS idx_mem_tree_entity_index_node
     ON mem_tree_entity_index(node_id);
 CREATE INDEX IF NOT EXISTS idx_mem_tree_entity_index_timestamp
     ON mem_tree_entity_index(timestamp_ms);
+
+-- Phase 3a (#709): summary trees / bucket-seal.
+-- `mem_tree_trees` tracks one tree per scope (source/topic/global).
+CREATE TABLE IF NOT EXISTS mem_tree_trees (
+    id                     TEXT PRIMARY KEY,
+    kind                   TEXT NOT NULL,
+    scope                  TEXT NOT NULL,
+    root_id                TEXT,
+    max_level              INTEGER NOT NULL DEFAULT 0,
+    status                 TEXT NOT NULL DEFAULT 'active',
+    created_at_ms          INTEGER NOT NULL,
+    last_sealed_at_ms      INTEGER
+);
+
+CREATE UNIQUE INDEX IF NOT EXISTS idx_mem_tree_trees_kind_scope
+    ON mem_tree_trees(kind, scope);
+CREATE INDEX IF NOT EXISTS idx_mem_tree_trees_status
+    ON mem_tree_trees(status);
+
+-- `mem_tree_summaries` holds sealed summary nodes. Immutable once written
+-- (Phase 3a). `deleted` is reserved for future archive cascades.
+CREATE TABLE IF NOT EXISTS mem_tree_summaries (
+    id                     TEXT PRIMARY KEY,
+    tree_id                TEXT NOT NULL,
+    tree_kind              TEXT NOT NULL,
+    level                  INTEGER NOT NULL,
+    parent_id              TEXT,
+    child_ids_json         TEXT NOT NULL DEFAULT '[]',
+    content                TEXT NOT NULL,
+    token_count            INTEGER NOT NULL,
+    entities_json          TEXT NOT NULL DEFAULT '[]',
+    topics_json            TEXT NOT NULL DEFAULT '[]',
+    time_range_start_ms    INTEGER NOT NULL,
+    time_range_end_ms      INTEGER NOT NULL,
+    score                  REAL NOT NULL DEFAULT 0.0,
+    sealed_at_ms           INTEGER NOT NULL,
+    deleted                INTEGER NOT NULL DEFAULT 0,
+    FOREIGN KEY (tree_id) REFERENCES mem_tree_trees(id)
+);
+
+CREATE INDEX IF NOT EXISTS idx_mem_tree_summaries_tree_level
+    ON mem_tree_summaries(tree_id, level);
+CREATE INDEX IF NOT EXISTS idx_mem_tree_summaries_parent
+    ON mem_tree_summaries(parent_id);
+CREATE INDEX IF NOT EXISTS idx_mem_tree_summaries_sealed_at
+    ON mem_tree_summaries(sealed_at_ms);
+CREATE INDEX IF NOT EXISTS idx_mem_tree_summaries_deleted
+    ON mem_tree_summaries(deleted);
+
+-- `mem_tree_buffers` holds the unsealed frontier per (tree, level). One row
+-- per active level per tree; deleted when the buffer seals (clears) in the
+-- same transaction as the new summary node row.
+CREATE TABLE IF NOT EXISTS mem_tree_buffers (
+    tree_id                TEXT NOT NULL,
+    level                  INTEGER NOT NULL,
+    item_ids_json          TEXT NOT NULL DEFAULT '[]',
+    token_sum              INTEGER NOT NULL DEFAULT 0,
+    oldest_at_ms           INTEGER,
+    updated_at_ms          INTEGER NOT NULL,
+    PRIMARY KEY (tree_id, level),
+    FOREIGN KEY (tree_id) REFERENCES mem_tree_trees(id)
+);
+
+CREATE INDEX IF NOT EXISTS idx_mem_tree_buffers_oldest
+    ON mem_tree_buffers(oldest_at_ms);
 ";
 
 /// Upsert a batch of chunks atomically.
@@ -355,6 +420,10 @@ pub(crate) fn with_connection<T>(
     // "no LLM signal available" by readers.
     add_column_if_missing(&conn, "mem_tree_score", "llm_importance", "REAL")?;
     add_column_if_missing(&conn, "mem_tree_score", "llm_importance_reason", "TEXT")?;
+    // Phase 3a (#709): parent-summary backlink on leaves. Populated when
+    // the L0 buffer seals into an L1 summary so traversal can walk
+    // leaf → parent without scanning `mem_tree_summaries.child_ids_json`.
+    add_column_if_missing(&conn, "mem_tree_chunks", "parent_summary_id", "TEXT")?;
     f(&conn)
 }