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

* feat(memory): LLM-based NER + importance signal with cheap-signals short-circuit (#708)

Adds an Ollama-backed entity extractor and an LLM-derived importance
score as a new signal in the admission gate. Builds on merged Phase 2
(#733) without changing its surface — additive only.

Why LLM (not GLiNER): zero new deps (reuses existing reqwest + async
infra), reuses the same Ollama setup openhuman uses for embeddings,
better per-entity quality scales with model choice, no native ONNX
runtime to ship. Latency cost (~100-300ms per chunk on a small model
like qwen2.5:0.5b) is amortised by the short-circuit.

The admission gate stays a hybrid - cheap deterministic signals
always run; the LLM is one signal among many, never the sole arbiter.
Backwards-compatible: with default SignalWeights the LLM weight is 0.0
and behaviour matches pre-LLM Phase 2 exactly.

What's new:

- src/openhuman/memory/tree/score/extract/llm.rs: LlmEntityExtractor
  implementing the existing EntityExtractor trait. Posts a structured
  JSON request to an Ollama-compatible /api/chat endpoint asking for
  NER + an importance rating in one call. Span recovery via string
  search; hallucinated entities (surface not in source text) dropped.
  Soft fallback: HTTP failures log a warn and return empty extraction.
- ExtractedEntities gains llm_importance (Option<f32>) +
  llm_importance_reason (Option<String>); merge() takes max importance.
- ScoreSignals gains llm_importance (f32, defaults to 0.0).
- SignalWeights gains llm_importance (default 0.0; with_llm_enabled()
  helper sets it to 2.0 - comparable to metadata/source weights).
- signals::combine_cheap_only(): variant that excludes the LLM signal,
  used for the short-circuit decision in score_chunk.
- ScoringConfig gains llm_extractor (Option<Arc<dyn EntityExtractor>>),
  definite_keep_threshold (default 0.85), definite_drop_threshold
  (default 0.15). New with_llm_extractor() constructor.
- score_chunk() pipeline:
  1. Always-on regex extraction
  2. Cheap signals + combine_cheap_only -> cheap_total
  3. If cheap_total >= definite_keep OR <= definite_drop: skip LLM
  4. Else (borderline band): run LLM extractor, merge results, recompute
  5. Final combine + admission gate against drop_threshold
- mem_tree_score table gains nullable llm_importance + llm_importance_reason
  columns via idempotent ALTER TABLE migration.
- ScoreRow + upsert/get persist the new column.

What's not changed:
- Existing JSON-RPC surface
- Default behaviour (LLM weight=0, no llm_extractor configured = same
  outputs as before)
- mem_tree_chunks / mem_tree_entity_index schemas

Tests added:
- LLM extractor: prompt construction, JSON parsing, hallucination
  drop, importance clamping, strict vs lenient unknown-kind handling
- Score pipeline: short-circuit on definite_keep/definite_drop skips
  LLM call (verified via FakeLlm call counter); borderline band
  consults LLM once; LLM failure falls back gracefully without erroring

LLM-NER work follows up on #708. Parent: #711.

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

* fix(memory): address CI failures and CodeRabbit review on LLM-NER (#708)

CI was red on PR #775 due to 10 E0063 errors — existing test-only struct
literals for ExtractedEntities / ScoreSignals / ScoreRow weren't updated
when the new optional llm_importance fields landed. Local cargo check
on lib-only passed; cargo check --tests caught them, as does CI.

CodeRabbit also flagged four semantic issues:

1. LlmEntityExtractor::extract() returned Err on HTTP transport, non-2xx
   status, and JSON parse failures. The soft-fallback contract (documented
   in the module header) only worked because score_chunk catches errors;
   any other caller got a hard error. Now split into extract_or_empty()
   which logs a warn on each failure mode and returns
   ExtractedEntities::default() unconditionally. The public extract()
   stays on the async-trait surface but never returns Err.

2. find_char_span() always searched from byte offset 0, so when the LLM
   returned the same surface twice (explicitly asked for duplicates in
   the prompt), both entities got (start=0, end=len) — same span, not
   distinct mentions. Added find_char_span_from(haystack, needle,
   byte_from, char_from) that resumes from a cursor. into_extracted_entities
   now tracks a per-surface (byte_after, char_after) cursor in a
   HashMap<String, (usize, u32)>, so duplicate mentions get distinct
   non-overlapping spans and over-claim (LLM returns 3 "Alice" when
   source has 2) drops the excess entries with a debug log.

3. combine() always included w.llm_importance in the denominator even
   when llm_importance=0.0 because the LLM didn't run. That artificially
   lowered the total on short-circuit paths. score_chunk now branches on
   llm_consulted: uses combine() when LLM ran (importance actually
   contributes), combine_cheap_only() when LLM was skipped or failed
   (denominator excludes the LLM weight). Observable in two new tests:
   short_circuit_reports_cheap_only_total verifies r.total ==
   combine_cheap_only(r.signals, weights) and strictly exceeds
   combine(r.signals, weights) when llm_importance=0; llm_consulted_
   reports_full_total verifies the opposite path.

4. Same issue from a different angle (CodeRabbit flagged both). Fixed
   by the same llm_consulted branch.

Test-literal fixes:
- signals/ops.rs: ScoreSignals literals pick up llm_importance field;
  make_entities helper uses ..Default::default() on ExtractedEntities.
- extract/types.rs: three test ExtractedEntities literals gain
  llm_importance: None, llm_importance_reason: None.
- resolver.rs: canonicalise_batch_preserves_spans literal gains the
  two fields.
- score/store.rs: sample_row gains signals.llm_importance and
  llm_importance_reason.

New tests (7 added):
- find_char_span_from_advances_past_prior_match
- find_char_span_from_returns_none_after_exhaustion
- find_char_span_from_preserves_utf8
- find_char_span_from_rejects_non_char_boundary
- into_extracted_entities_gives_distinct_spans_to_duplicate_mentions
- into_extracted_entities_drops_extra_duplicate_when_source_only_has_one
- extract_soft_fallback_on_unreachable_endpoint (transport failure →
  empty extraction, not Err)
- short_circuit_reports_cheap_only_total
- llm_consulted_reports_full_total

cargo check --lib clean; cargo fmt clean. The integration-test rmeta
errors visible locally are stale-metadata artifacts from incremental
builds with prior struct shapes; CI's clean build resolves them and
tests/*.rs files do not construct any of the touched types.

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

* 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>

* style(memory): cargo fmt on source_tree module (#709)

CI rustfmt caught a few multi-line function signatures and import
groups that rustfmt wants collapsed / reordered. No behaviour change.

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

* fix(memory): address CodeRabbit review on source-tree (#709)

Three correctness fixes from CodeRabbit on PR #789, plus a design
clarification for the InertSummariser. False-positive findings
(schema-out-of-sync, rustfmt — already in b7395a8) are not addressed
here because they are not real issues.

## Changes

### Idempotent buffer append (`bucket_seal.rs`)
`append_to_buffer` commits before the async seal/cascade runs, so a
retry after a failed seal (summariser timeout, crash, transient ingest
error) would re-push the same leaf and double-count tokens. Added a
dedup check on `item_ids` so the append is a true no-op for
already-buffered chunks.

### InertSummariser: honest stub, no entity propagation (`summariser/inert.rs`)
Summary-level entities and topics are **LLM-derived from the summary's
own synthesised content** by design, not a mechanical union of
children's labels. The networked summariser (future) does its own
extraction on its output. The inert fallback has no NER, so it emits
empty entities/topics — an honest stub rather than a misleading union.
Removed `union_preserve_order` helper and updated the corresponding
test.

### Forward-compat entity index on seal (`bucket_seal.rs`, `score/store.rs`)
Added `index_summary_entity_ids_tx` — a summary-specific indexer that
takes canonical ids only (matching the `Vec<String>` shape of
`SummaryOutput.entities`) and writes them into `mem_tree_entity_index`
with `node_kind='summary'`, placeholder `entity_kind`/`surface` (both
set to the canonical id), and the summary's score. Called from
`seal_one_level`'s write transaction. No-op today (InertSummariser
emits empty), but wired so the Ollama summariser lands as a
drop-in without touching bucket-seal.

### UNIQUE race in `get_or_create_source_tree` (`registry.rs`)
Two concurrent callers for the same scope could both pass the lookup
and then race on `insert_tree`; the loser got a UNIQUE violation
bubbled as an error instead of the pre-existing row. Now catches
`ConstraintViolation` (and the message-based fallback for chained
errors), re-queries by scope, and returns the winning row. New test
`get_or_create_recovers_from_unique_race` covers both the recovery
path and the `is_unique_violation` detector.

## Tests

161 passed / 0 failed across `memory::tree` (+1 new test vs the
prior baseline of 160). No regressions.

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

* style(memory): cargo fmt on source_tree::registry (#709)

CI's rustfmt check flagged the race-recovery block in
get_or_create_source_tree — rustfmt prefers chaining
`?.ok_or_else(...)` on one line and wrapping the assert! macro.
No behavior change.

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

* fix(memory): index_summary_entity_ids_tx writes parseable entity_kind (#709)

CodeRabbit on PR #789 caught a bug in the summary entity indexer I
added for Phase 3a. The helper was writing the full canonical id
(e.g. "email:alice@example.com") into the `entity_kind` column instead
of the kind prefix. `lookup_entity()` later runs `EntityKind::parse()`
on that column and would fail with `unknown EntityKind` the first time
a result set mixed leaf and summary hits — blocking any Phase 4 read
that needed summary-level entity resolution.

Fix: split the canonical id on `:` and store the prefix only. Falls
back to the full id with a `warn!` for malformed ids that lack `:`,
so bad data surfaces rather than silently stays latent.

Added regression test `summary_entity_index_kind_is_parseable` that:
- Indexes a leaf entity row (the existing happy path)
- Indexes two summary entity rows via index_summary_entity_ids_tx
- Runs lookup_entity for email, hashtag, and a cross-kind id
- Asserts each row comes back with a correctly-parsed EntityKind

Before the fix, the lookup_entity calls in this test would fail with
a FromSqlConversionFailure on the summary row's entity_kind column.
After the fix, mixed leaf+summary lookups round-trip cleanly.

Tests: 162 passed / 0 failed in memory::tree (+1 vs 161 baseline).

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

* feat(memory): promote extracted topics to canonical entities (#709)

Addresses a design gap surfaced during PR #789 review: the Phase 3
plan promised topic trees would scope to "entities / topics" but the
implementation routed on canonical_entities only. A chunk saying
"Phoenix migration ships Friday" produced `r.extracted.topics =
["phoenix", "migration"]` but `canonical_entities = []` — meaning no
topic tree named Phoenix would ever spawn.

Fix (Shape A from the review discussion): extend the canonical entity
stream with topic rows. No new tables, no new routing path, no new
hotness counters.

## Changes

- `score/extract/types.rs`: add `EntityKind::Topic` with as_str/parse
  round-trip. `non_exhaustive` enum so this is an additive change.
- `score/resolver.rs::canonicalise`: after emitting canonical entities
  from `extracted.entities`, also emit one per `extracted.topics`
  entry with `kind: Topic`, `canonical_id: "topic:<lowercased>"`.
  Span fields are 0 (topics are chunk-level, not substring-scoped).
  Dedups same-label topics internally; keeps `hashtag:launch` and
  `topic:launch` as separate entities by design.

## Downstream effect (zero-touch)

- `score::store::index_entities_tx` indexes topic entities the same
  way as email entities — rows land in `mem_tree_entity_index` with
  `entity_kind = "topic"`.
- Phase 3c routing iterates `canonical_entities` — topics now trigger
  topic-tree routing automatically. A chunk about "phoenix" starts
  accumulating hotness for `topic:phoenix`; once the threshold
  crosses, `get_or_create_topic_tree("topic:phoenix")` materialises
  a dedicated topic tree. Backfill via `lookup_entity("topic:phoenix")`
  then hydrates historic leaves.
- Phase 4 retrieval can filter `WHERE entity_kind = 'topic'` to
  surface themes without mixing in people/emails.

## Tests

Five new tests in `resolver.rs`:
- topics → canonical entities with `kind: Topic`
- lowercase normalisation + dedup on label
- hashtag + topic with the same label coexist (different kind prefix)
- entities emitted first, topics appended
- `EntityKind::Topic` round-trips through `parse` / `as_str`

Broader memory::tree: 167 passed / 0 failed (162 baseline + 5 new).
No regressions.

## Caveats for follow-up

1. Topic noise — LLM-extracted themes are softer signal than regex
   entities. Hotness threshold (TOPIC_CREATION_THRESHOLD=10.0) gates
   tree creation, so transient topics won't spawn trees.
2. Topic/hashtag duplication — "Phoenix #phoenix" creates both
   `topic:phoenix` and `hashtag:phoenix` trees. Tolerable (kinds are
   semantically distinct) but a dedup policy may be worth revisiting
   in a Phase 4 retrieval follow-up.
3. Topic normalisation is lowercase+trim only. Plurals / stemming
   are deferred.

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

---------

Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Co-authored-by: Steven Enamakel <31011319+senamakel@users.noreply.github.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/score/extract/types.rs

@@ -33,6 +33,13 @@ pub enum EntityKind {
     Event,
     Product,
     Misc,
+    // Thematic — scorer-surfaced topics (hashtag-like short phrases or
+    // LLM-extracted themes). Promoted into the canonical entity stream
+    // by the resolver so Phase 3c topic trees can route on themes the
+    // same way they route on people/orgs. A chunk saying "Phoenix
+    // migration ships Friday" emits `topic:phoenix` and `topic:migration`
+    // in addition to any emails/hashtags the mechanical extractors find.
+    Topic,
 }
 
 impl EntityKind {
@@ -48,6 +55,7 @@ impl EntityKind {
             Self::Event => "event",
             Self::Product => "product",
             Self::Misc => "misc",
+            Self::Topic => "topic",
         }
     }
 
@@ -63,6 +71,7 @@ impl EntityKind {
             "event" => Ok(Self::Event),
             "product" => Ok(Self::Product),
             "misc" => Ok(Self::Misc),
+            "topic" => Ok(Self::Topic),
             other => Err(format!("unknown entity kind: {other}")),
         }
     }

src/openhuman/memory/tree/score/resolver.rs

@@ -28,8 +28,15 @@ pub struct CanonicalEntity {
 /// Same surface form (after normalisation) → same `canonical_id` regardless
 /// of how many times it appears in a chunk. Preserves source spans by
 /// emitting one [`CanonicalEntity`] per occurrence.
+///
+/// Extracted **topics** are also promoted into the canonical stream under
+/// [`EntityKind::Topic`] so downstream routing (Phase 3c topic trees) can
+/// treat themes as first-class scope alongside people/orgs. Topics have no
+/// source span (they're derived from the whole chunk, not a specific
+/// substring), so `span_start` / `span_end` are both `0` for topic rows —
+/// readers should key on `kind` instead of span when span-awareness matters.
 pub fn canonicalise(extracted: &ExtractedEntities) -> Vec<CanonicalEntity> {
-    extracted
+    let mut out: Vec<CanonicalEntity> = extracted
         .entities
         .iter()
         .map(|e| CanonicalEntity {
@@ -40,7 +47,35 @@ pub fn canonicalise(extracted: &ExtractedEntities) -> Vec<CanonicalEntity> {
             span_end: e.span_end,
             score: e.score,
         })
-        .collect()
+        .collect();
+
+    // Promote topics. Dedup against the entities we already emitted so a
+    // hashtag like `#launch` and a topic label `"launch"` don't both land
+    // as the same canonical id with the same kind — the hashtag keeps its
+    // Hashtag kind, the topic gets Topic kind, and `canonical_id_for`
+    // makes them distinguishable: `hashtag:launch` vs `topic:launch`.
+    for topic in &extracted.topics {
+        let canonical_id = canonical_id_for(EntityKind::Topic, &topic.label);
+        // Dedup within the topic set in case the scorer produces the same
+        // label twice (LLM + regex overlap). Entities under other kinds
+        // aren't dedup targets — `topic:launch` and `hashtag:launch` are
+        // intentionally separate.
+        if out
+            .iter()
+            .any(|e| e.kind == EntityKind::Topic && e.canonical_id == canonical_id)
+        {
+            continue;
+        }
+        out.push(CanonicalEntity {
+            canonical_id,
+            kind: EntityKind::Topic,
+            surface: topic.label.clone(),
+            span_start: 0,
+            span_end: 0,
+            score: topic.score,
+        });
+    }
+    out
 }
 
 /// Canonical id form per kind. Deterministic so the same surface always
@@ -135,4 +170,96 @@ mod tests {
             canonical_id_for(EntityKind::Person, "alice")
         );
     }
+
+    // ── Topic canonicalisation (#709 / Phase 3c topic-tree scope) ────
+
+    use crate::openhuman::memory::tree::score::extract::ExtractedTopic;
+
+    fn topic(label: &str, score: f32) -> ExtractedTopic {
+        ExtractedTopic {
+            label: label.to_string(),
+            score,
+        }
+    }
+
+    #[test]
+    fn topics_are_promoted_to_canonical_entities() {
+        let ex = ExtractedEntities {
+            entities: vec![],
+            topics: vec![topic("phoenix", 0.72), topic("migration", 0.60)],
+            llm_importance: None,
+            llm_importance_reason: None,
+        };
+        let out = canonicalise(&ex);
+        assert_eq!(out.len(), 2);
+        assert_eq!(out[0].kind, EntityKind::Topic);
+        assert_eq!(out[0].canonical_id, "topic:phoenix");
+        assert!((out[0].score - 0.72).abs() < 1e-6);
+        assert_eq!(out[1].canonical_id, "topic:migration");
+    }
+
+    #[test]
+    fn topic_canonicalisation_lowercases() {
+        let ex = ExtractedEntities {
+            entities: vec![],
+            topics: vec![topic("Phoenix", 1.0), topic("PHOENIX", 0.5)],
+            llm_importance: None,
+            llm_importance_reason: None,
+        };
+        let out = canonicalise(&ex);
+        // Both normalise to "topic:phoenix" — second occurrence is deduped.
+        assert_eq!(out.len(), 1);
+        assert_eq!(out[0].canonical_id, "topic:phoenix");
+        // First-seen surface is preserved.
+        assert_eq!(out[0].surface, "Phoenix");
+    }
+
+    #[test]
+    fn hashtag_and_topic_with_same_label_coexist() {
+        // "#launch" regex → EntityKind::Hashtag, LLM theme "launch" → Topic.
+        // They stay as two distinct canonical entities — different kind,
+        // different canonical_id prefix.
+        let ex = ExtractedEntities {
+            entities: vec![ExtractedEntity {
+                kind: EntityKind::Hashtag,
+                text: "launch".into(),
+                span_start: 0,
+                span_end: 6,
+                score: 1.0,
+            }],
+            topics: vec![topic("launch", 0.8)],
+            llm_importance: None,
+            llm_importance_reason: None,
+        };
+        let out = canonicalise(&ex);
+        assert_eq!(out.len(), 2);
+        assert_eq!(out[0].kind, EntityKind::Hashtag);
+        assert_eq!(out[0].canonical_id, "hashtag:launch");
+        assert_eq!(out[1].kind, EntityKind::Topic);
+        assert_eq!(out[1].canonical_id, "topic:launch");
+    }
+
+    #[test]
+    fn canonicalise_mixes_entities_and_topics_in_order() {
+        // Entities come first, topics appended after — downstream callers
+        // (e.g. routing) can rely on this ordering if they ever need it.
+        let ex = ExtractedEntities {
+            entities: vec![entity(EntityKind::Email, "alice@example.com")],
+            topics: vec![topic("phoenix", 0.7)],
+            llm_importance: None,
+            llm_importance_reason: None,
+        };
+        let out = canonicalise(&ex);
+        assert_eq!(out.len(), 2);
+        assert_eq!(out[0].kind, EntityKind::Email);
+        assert_eq!(out[1].kind, EntityKind::Topic);
+    }
+
+    #[test]
+    fn topic_entity_kind_round_trips_through_parse() {
+        // Defence in depth: ensure the new Topic variant survives the
+        // round-trip used by mem_tree_entity_index on read.
+        assert_eq!(EntityKind::parse("topic"), Ok(EntityKind::Topic));
+        assert_eq!(EntityKind::Topic.as_str(), "topic");
+    }
 }

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

@@ -223,6 +223,65 @@ pub(crate) fn clear_entity_index_for_node_tx(tx: &Transaction<'_>, node_id: &str
     Ok(n)
 }
 
+/// Index summary-node entities by canonical id only. Summary-level entity
+/// metadata is LLM-derived (Phase 3a #709) — the summariser emits a
+/// curated list of canonical ids without per-occurrence span/surface data.
+///
+/// Writes the kind prefix (everything before the first `:`) into the
+/// `entity_kind` column so [`lookup_entity`]'s `EntityKind::parse()` keeps
+/// round-tripping on summary rows. `surface` stores the full canonical id
+/// as a stable placeholder — at the summary level we have no per-occurrence
+/// span to recover, and the id is always unique. The summary's score is
+/// reused for each of its entities.
+///
+/// Callers should prefer the regular [`index_entities_tx`] for leaves,
+/// where span/surface are meaningful.
+pub(crate) fn index_summary_entity_ids_tx(
+    tx: &Transaction<'_>,
+    entity_ids: &[String],
+    node_id: &str,
+    score: f32,
+    timestamp_ms: i64,
+    tree_id: Option<&str>,
+) -> Result<usize> {
+    if entity_ids.is_empty() {
+        return Ok(0);
+    }
+    let mut stmt = tx.prepare(
+        "INSERT OR REPLACE INTO mem_tree_entity_index (
+            entity_id, node_id, node_kind, entity_kind, surface,
+            score, timestamp_ms, tree_id
+         ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)",
+    )?;
+    for canonical_id in entity_ids {
+        // Canonical ids follow Phase 2's "<kind>:<value>" convention.
+        // Without this split, `entity_kind` would hold the full id and
+        // `lookup_entity`'s `EntityKind::parse()` would fail at read time,
+        // poisoning any mixed leaf/summary lookup.
+        let entity_kind = match canonical_id.split_once(':') {
+            Some((kind, _)) => kind,
+            None => {
+                log::warn!(
+                    "[memory_tree::score::store] summary entity id missing ':' — \
+                     storing as-is: {canonical_id}"
+                );
+                canonical_id.as_str()
+            }
+        };
+        stmt.execute(params![
+            canonical_id,
+            node_id,
+            "summary",
+            entity_kind,
+            canonical_id,
+            score,
+            timestamp_ms,
+            tree_id,
+        ])?;
+    }
+    Ok(entity_ids.len())
+}
+
 pub(crate) fn index_entities_tx(
     tx: &Transaction<'_>,
     entities: &[CanonicalEntity],
@@ -485,4 +544,58 @@ mod tests {
         let hits = lookup_entity(&cfg, "email:alice", Some(2)).unwrap();
         assert_eq!(hits.len(), 2);
     }
+
+    /// Regression: `index_summary_entity_ids_tx` must write a parseable
+    /// `entity_kind` (the "<kind>" prefix before `:`) so `lookup_entity`
+    /// can still round-trip rows through `EntityKind::parse`. Earlier code
+    /// stored the full canonical id, which poisoned lookups mixing leaf
+    /// and summary hits. See PR #789 CodeRabbit review.
+    #[test]
+    fn summary_entity_index_kind_is_parseable() {
+        use crate::openhuman::memory::tree::store::with_connection;
+
+        let (_tmp, cfg) = test_config();
+
+        // Seed a leaf hit so lookup_entity has something leafy to mix
+        // with the summary hit — this reproduces the mixed-row crash.
+        let leaf_entity = sample_entity("alice");
+        index_entity(&cfg, &leaf_entity, "leaf-1", "leaf", 1000, Some("tree-1")).unwrap();
+
+        // Write a summary row via the tx helper under test.
+        with_connection(&cfg, |conn| {
+            let tx = conn.unchecked_transaction()?;
+            let n = index_summary_entity_ids_tx(
+                &tx,
+                &["email:alice@example.com".into(), "hashtag:launch-q2".into()],
+                "summary-1",
+                0.84,
+                2000,
+                Some("tree-1"),
+            )?;
+            assert_eq!(n, 2);
+            tx.commit()?;
+            Ok(())
+        })
+        .unwrap();
+
+        // Before the fix: lookup_entity would fail on the summary row
+        // because entity_kind was "email:alice@example.com" and
+        // EntityKind::parse rejects it. After the fix, the column stores
+        // "email" and the lookup succeeds with both rows.
+        let hits = lookup_entity(&cfg, "email:alice@example.com", None).unwrap();
+        assert_eq!(hits.len(), 1, "summary row should be discoverable");
+        assert_eq!(hits[0].node_id, "summary-1");
+        assert_eq!(hits[0].node_kind, "summary");
+        assert_eq!(hits[0].entity_kind, EntityKind::Email);
+
+        // Hashtag row parses as its own kind too.
+        let hits = lookup_entity(&cfg, "hashtag:launch-q2", None).unwrap();
+        assert_eq!(hits.len(), 1);
+        assert_eq!(hits[0].entity_kind, EntityKind::Hashtag);
+
+        // Mixing leaf + summary entity ids in one lookup also parses cleanly.
+        let hits = lookup_entity(&cfg, "email:alice", None).unwrap();
+        assert_eq!(hits.len(), 1);
+        assert_eq!(hits[0].entity_kind, EntityKind::Email);
+    }
 }

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

@@ -0,0 +1,566 @@
+//! Append + cascade-seal for summary trees (#709).
+//!
+//! `append_leaf` pushes a persisted chunk into the L0 buffer of a tree. If
+//! the buffer's running `token_sum` crosses `TOKEN_BUDGET`, the buffer
+//! seals into a level-1 summary, its items move into the summary's
+//! `child_ids`, the buffer clears, and the new summary id is queued at
+//! level 2. The cascade continues upward until a buffer stays under the
+//! token budget.
+//!
+//! Concurrency: Phase 3a assumes a single-process SQLite workspace. All
+//! writes in one seal step run in a single transaction; the async
+//! summariser call happens outside any open transaction so a slow LLM
+//! doesn't hold DB locks. Callers should serialise `append_leaf` per
+//! tree — ingest achieves this by processing one batch's chunks
+//! sequentially inside its `persist` task. Blocking SQLite calls inside
+//! this async function are acceptable for Phase 3a because the Inert
+//! summariser does no real I/O; when a networked summariser lands, wrap
+//! DB calls in `tokio::task::spawn_blocking` to keep the runtime healthy.
+
+use anyhow::{Context, Result};
+use chrono::{DateTime, Utc};
+use rusqlite::Transaction;
+
+use crate::openhuman::config::Config;
+use crate::openhuman::memory::tree::source_tree::registry::new_summary_id;
+use crate::openhuman::memory::tree::source_tree::store;
+use crate::openhuman::memory::tree::source_tree::summariser::{
+    Summariser, SummaryContext, SummaryInput,
+};
+use crate::openhuman::memory::tree::source_tree::types::{
+    Buffer, SummaryNode, Tree, TreeKind, TOKEN_BUDGET,
+};
+use crate::openhuman::memory::tree::store::with_connection;
+
+/// Hard cap on cascade depth — prevents runaway loops if token accounting
+/// ever slips. 32 levels at even a 2x fan-in is more than enough for any
+/// realistic source.
+const MAX_CASCADE_DEPTH: u32 = 32;
+
+/// A single leaf being appended to an L0 buffer.
+#[derive(Clone, Debug)]
+pub struct LeafRef {
+    pub chunk_id: String,
+    pub token_count: u32,
+    pub timestamp: DateTime<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)?;
+        // Idempotent on (tree_id, level, item_id): a retry after a failed
+        // cascade (step 2 of append_leaf) is a no-op, so duplicated children
+        // and double-counted tokens can't slip into the buffer. oldest_at
+        // stays on first-seen.
+        if buf.item_ids.iter().any(|existing| existing == item_id) {
+            log::debug!(
+                "[source_tree::bucket_seal] append_to_buffer: {item_id} already in buffer \
+                 tree_id={tree_id} level={level} — no-op"
+            );
+            return Ok(());
+        }
+        buf.item_ids.push(item_id.to_string());
+        buf.token_sum = buf.token_sum.saturating_add(token_delta);
+        buf.oldest_at = match buf.oldest_at {
+            Some(existing) => Some(existing.min(item_ts)),
+            None => Some(item_ts),
+        };
+        store::upsert_buffer_tx(&tx, &buf)?;
+        tx.commit()?;
+        Ok(())
+    })
+}
+
+async fn cascade_seals(
+    config: &Config,
+    tree: &Tree,
+    summariser: &dyn Summariser,
+) -> Result<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)?;
+        // Forward-compat: index any entities the summariser emitted into
+        // `mem_tree_entity_index` so Phase 4 retrieval can resolve
+        // "summaries mentioning Alice" via the same inverted index as
+        // leaves. No-op under InertSummariser (entities is empty by
+        // design — see summariser/inert.rs doc); becomes active once the
+        // Ollama summariser lands and emits curated canonical ids.
+        crate::openhuman::memory::tree::score::store::index_summary_entity_ids_tx(
+            &tx,
+            &node.entities,
+            &node.id,
+            node.score,
+            now.timestamp_millis(),
+            Some(&tree_id),
+        )?;
+        // Backlink children → new parent so leaf/parent traversal is a
+        // single-row lookup in Phase 4. Skipped for levels already bound
+        // to a parent (shouldn't happen — a child seals at most once).
+        for child_id in &node.child_ids {
+            if level == 0 {
+                tx.execute(
+                    "UPDATE mem_tree_chunks
+                        SET parent_summary_id = ?1
+                      WHERE id = ?2 AND parent_summary_id IS NULL",
+                    rusqlite::params![&summary_id_for_closure, child_id],
+                )
+                .context("Failed to backlink chunk to parent summary")?;
+            } else {
+                tx.execute(
+                    "UPDATE mem_tree_summaries
+                        SET parent_id = ?1
+                      WHERE id = ?2 AND parent_id IS NULL",
+                    rusqlite::params![&summary_id_for_closure, child_id],
+                )
+                .context("Failed to backlink summary to parent summary")?;
+            }
+        }
+        store::clear_buffer_tx(&tx, &tree_id, level)?;
+
+        // Append to parent buffer.
+        let mut parent = store::get_buffer_conn(&tx, &tree_id, target_level_for_closure)?;
+        parent.item_ids.push(summary_id_for_closure.clone());
+        parent.token_sum = parent.token_sum.saturating_add(node.token_count as i64);
+        parent.oldest_at = match parent.oldest_at {
+            Some(existing) => Some(existing.min(time_range_start)),
+            None => Some(time_range_start),
+        };
+        store::upsert_buffer_tx(&tx, &parent)?;
+
+        // Update tree root / max_level if we just climbed.
+        if target_level_for_closure > current_max {
+            store::update_tree_after_seal_tx(
+                &tx,
+                &tree_id,
+                &summary_id_for_closure,
+                target_level_for_closure,
+                now,
+            )?;
+        } else {
+            // Same max level — still refresh last_sealed_at via same helper
+            // but keep existing root intact. Root tracking at the same
+            // level is resolved in Phase 4 retrieval.
+            refresh_last_sealed_tx(&tx, &tree_id, now)?;
+        }
+
+        tx.commit()?;
+        Ok(())
+    })?;
+
+    log::info!(
+        "[source_tree::bucket_seal] sealed tree_id={} level={}→{} summary_id={} children={}",
+        tree.id,
+        level,
+        target_level,
+        summary_id,
+        buf.item_ids.len()
+    );
+
+    Ok(summary_id)
+}
+
+fn refresh_last_sealed_tx(
+    tx: &Transaction<'_>,
+    tree_id: &str,
+    sealed_at: DateTime<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,190 @@
+//! Time-based buffer flush for source trees (#709).
+//!
+//! The bucket-seal path only fires when a buffer crosses `TOKEN_BUDGET`.
+//! Low-volume sources (e.g. an email account with two threads a week) can
+//! otherwise leave leaves parked in the L0 buffer indefinitely, which
+//! hurts recall. `flush_stale_buffers` force-seals any buffer whose
+//! `oldest_at` is older than `max_age`, regardless of token count.
+//!
+//! This is meant to run on a cadence (e.g. daily cron). Phase 3a ships
+//! the primitive; wiring into a scheduler is not required for merge.
+
+use anyhow::Result;
+use chrono::{DateTime, Duration, Utc};
+
+use crate::openhuman::config::Config;
+use crate::openhuman::memory::tree::source_tree::bucket_seal::cascade_all_from;
+use crate::openhuman::memory::tree::source_tree::store;
+use crate::openhuman::memory::tree::source_tree::summariser::Summariser;
+use crate::openhuman::memory::tree::source_tree::types::DEFAULT_FLUSH_AGE_SECS;
+
+/// Seal every buffer whose oldest item is older than `max_age`. Returns
+/// the number of individual seal calls (not trees) that fired. When the
+/// same tree has multiple stale levels they're each sealed in order.
+pub async fn flush_stale_buffers(
+    config: &Config,
+    max_age: Duration,
+    summariser: &dyn Summariser,
+) -> Result<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,172 @@
+//! Tree registry — get-or-create for source trees (#709).
+//!
+//! The registry is the entry point for the ingest path to look up the
+//! tree for a given (kind, scope). Phase 3a only touches source trees;
+//! topic / global trees will reuse the same `(kind, scope)` convention
+//! in Phases 3b / 3c.
+
+use anyhow::Result;
+use chrono::Utc;
+use uuid::Uuid;
+
+use crate::openhuman::config::Config;
+use crate::openhuman::memory::tree::source_tree::store;
+use crate::openhuman::memory::tree::source_tree::types::{Tree, TreeKind, TreeStatus};
+
+/// Look up the source tree for `scope`, or create a new one.
+///
+/// Scope format convention (Phase 3a): use the ingested chunk's
+/// `metadata.source_id` verbatim, so re-ingesting the same Slack channel
+/// or Gmail account keeps appending to the same tree.
+pub fn get_or_create_source_tree(config: &Config, scope: &str) -> Result<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,
+    };
+    match store::insert_tree(config, &tree) {
+        Ok(()) => {
+            log::info!(
+                "[source_tree::registry] created source tree id={} scope={}",
+                tree.id,
+                scope
+            );
+            Ok(tree)
+        }
+        Err(err) if is_unique_violation(&err) => {
+            // Race: another caller created a tree for the same scope
+            // between our initial lookup and this insert. UNIQUE(kind,
+            // scope) rejected our row; re-query and return the winner.
+            log::debug!(
+                "[source_tree::registry] UNIQUE race for scope={} — re-querying",
+                scope
+            );
+            store::get_tree_by_scope(config, TreeKind::Source, scope)?.ok_or_else(|| {
+                anyhow::anyhow!(
+                    "UNIQUE violation on insert but no row found on re-query for scope {scope}"
+                )
+            })
+        }
+        Err(err) => Err(err),
+    }
+}
+
+/// Return true if `err` represents a SQLite UNIQUE constraint violation.
+/// Matches both the anyhow-wrapped rusqlite error text and the raw SQLite
+/// error codes in case the wrapping chain is shorter.
+fn is_unique_violation(err: &anyhow::Error) -> bool {
+    if let Some(rusqlite_err) = err.downcast_ref::<rusqlite::Error>() {
+        if let rusqlite::Error::SqliteFailure(sqlite_err, _) = rusqlite_err {
+            return sqlite_err.code == rusqlite::ErrorCode::ConstraintViolation;
+        }
+    }
+    // Fallback for chained/wrapped errors: scan the rendered message.
+    let msg = format!("{err:#}");
+    msg.contains("UNIQUE constraint failed")
+}
+
+fn new_tree_id(kind: TreeKind) -> String {
+    format!("{}:{}", kind.as_str(), Uuid::new_v4())
+}
+
+/// Public id generator for summary nodes — exported so `bucket_seal` can
+/// share the same format (kept separate for readability; both use UUID v4
+/// suffixes to keep ids short but unambiguous).
+pub fn new_summary_id(level: u32) -> String {
+    format!("summary:L{}:{}", level, Uuid::new_v4())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use tempfile::TempDir;
+
+    fn test_config() -> (TempDir, Config) {
+        let tmp = TempDir::new().unwrap();
+        let mut cfg = Config::default();
+        cfg.workspace_dir = tmp.path().to_path_buf();
+        (tmp, cfg)
+    }
+
+    #[test]
+    fn get_or_create_is_idempotent_on_scope() {
+        let (_tmp, cfg) = test_config();
+        let first = get_or_create_source_tree(&cfg, "slack:#eng").unwrap();
+        let second = get_or_create_source_tree(&cfg, "slack:#eng").unwrap();
+        assert_eq!(first.id, second.id);
+        assert_eq!(first.kind, TreeKind::Source);
+        assert_eq!(first.status, TreeStatus::Active);
+    }
+
+    #[test]
+    fn different_scopes_yield_different_trees() {
+        let (_tmp, cfg) = test_config();
+        let a = get_or_create_source_tree(&cfg, "slack:#eng").unwrap();
+        let b = get_or_create_source_tree(&cfg, "gmail:user@example.com").unwrap();
+        assert_ne!(a.id, b.id);
+        assert_ne!(a.scope, b.scope);
+    }
+
+    #[test]
+    fn tree_id_has_expected_prefix() {
+        let id = new_tree_id(TreeKind::Source);
+        assert!(id.starts_with("source:"));
+        let sum_id = new_summary_id(3);
+        assert!(sum_id.starts_with("summary:L3:"));
+    }
+
+    #[test]
+    fn get_or_create_recovers_from_unique_race() {
+        // Simulate the race by pre-inserting a tree under the same scope
+        // with a different id. `get_or_create` must re-query and return
+        // the pre-existing row, not bubble the UNIQUE error.
+        let (_tmp, cfg) = test_config();
+        let pre_existing = Tree {
+            id: "source:preexisting".into(),
+            kind: TreeKind::Source,
+            scope: "slack:#eng".into(),
+            root_id: None,
+            max_level: 0,
+            status: TreeStatus::Active,
+            created_at: Utc::now(),
+            last_sealed_at: None,
+        };
+        store::insert_tree(&cfg, &pre_existing).unwrap();
+
+        // First call finds it via get_tree_by_scope (happy path — no race
+        // triggered here). To hit the race branch we need a caller that
+        // skips the lookup and goes straight to insert with a fresh id.
+        // Simplest proxy: call get_or_create twice from this test thread;
+        // the first creates, the second's UNIQUE would fire if the
+        // lookup was ever elided. Instead we cover the race path directly
+        // via `is_unique_violation` on a synthesised insert failure below.
+        let got = get_or_create_source_tree(&cfg, "slack:#eng").unwrap();
+        assert_eq!(got.id, "source:preexisting");
+
+        // Direct coverage: a second insert with a different id for the
+        // same scope must surface as UNIQUE and be detected.
+        let dup = Tree {
+            id: "source:would-collide".into(),
+            ..pre_existing.clone()
+        };
+        let err = store::insert_tree(&cfg, &dup).unwrap_err();
+        assert!(
+            is_unique_violation(&err),
+            "expected UNIQUE violation, got: {err:#}"
+        );
+    }
+}

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

@@ -0,0 +1,594 @@
+//! SQLite-backed persistence for Phase 3a summary trees (#709).
+//!
+//! Three tables (schema lives in the sibling `tree::store::SCHEMA`):
+//! - `mem_tree_trees`      — one row per tree (kind, scope, root, max_level)
+//! - `mem_tree_summaries`  — one row per sealed summary node (immutable)
+//! - `mem_tree_buffers`    — one row per unsealed frontier `(tree_id, level)`
+//!
+//! All timestamps are stored as milliseconds since the Unix epoch so we
+//! share the epoch convention with `mem_tree_chunks`. Writes are serialised
+//! through the sibling `tree::store::with_connection` so we inherit its
+//! busy-timeout, WAL, and schema-init behaviour.
+
+use anyhow::{Context, Result};
+use chrono::{DateTime, TimeZone, Utc};
+use rusqlite::{params, Connection, OptionalExtension, Transaction};
+
+use crate::openhuman::config::Config;
+use crate::openhuman::memory::tree::source_tree::types::{
+    Buffer, SummaryNode, Tree, TreeKind, TreeStatus,
+};
+use crate::openhuman::memory::tree::store::with_connection;
+
+fn ms_to_utc(ms: i64) -> rusqlite::Result<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,176 @@
+//! Deterministic fallback summariser (#709).
+//!
+//! `InertSummariser` concatenates each input's content, separated by a
+//! blank line, and hard-truncates to `ctx.token_budget`. Entities and
+//! topics are **intentionally empty**: per design, summary-level entity /
+//! topic metadata is derived by the LLM summariser from the summary's own
+//! synthesised content (not by mechanically unioning children's labels).
+//! Until the networked summariser lands, inert-sealed summaries have no
+//! entity index rows — an honest stub. The goal of this fallback is not
+//! metadata fidelity; it's a stable, dependency-free baseline so tree
+//! mechanics (sealing, cascade, roots) can be tested without an LLM.
+
+use anyhow::Result;
+use async_trait::async_trait;
+
+use crate::openhuman::memory::tree::source_tree::summariser::{
+    Summariser, SummaryContext, SummaryInput, SummaryOutput,
+};
+use crate::openhuman::memory::tree::types::approx_token_count;
+
+/// Default prefix applied to each contribution in the joined body. Keeps
+/// provenance visible to a human reading the raw summary.
+const PROVENANCE_PREFIX: &str = "— ";
+
+pub struct InertSummariser;
+
+impl InertSummariser {
+    pub fn new() -> Self {
+        Self
+    }
+}
+
+impl Default for InertSummariser {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+#[async_trait]
+impl Summariser for InertSummariser {
+    async fn summarise(
+        &self,
+        inputs: &[SummaryInput],
+        ctx: &SummaryContext<'_>,
+    ) -> Result<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);
+
+        log::debug!(
+            "[source_tree::summariser::inert] sealed tree_id={} level={} inputs={} tokens={} \
+             entities=0 topics=0 (honest stub — LLM summariser derives these)",
+            ctx.tree_id,
+            ctx.target_level,
+            inputs.len(),
+            token_count
+        );
+
+        Ok(SummaryOutput {
+            content,
+            token_count,
+            entities: Vec::new(),
+            topics: Vec::new(),
+        })
+    }
+}
+
+/// Truncate `text` to fit within `budget` approximate tokens. Returns the
+/// (possibly truncated) body and its recomputed token count. Truncation is
+/// done on character boundaries — `approx_token_count` assumes ~4 chars
+/// per token so we clamp character length to `budget * 4`.
+fn truncate_to_budget(text: &str, budget: u32) -> (String, u32) {
+    let initial = approx_token_count(text);
+    if initial <= budget {
+        return (text.to_string(), initial);
+    }
+    // Character ceiling derived from the same ~4 chars/token heuristic.
+    let char_ceiling = (budget as usize).saturating_mul(4);
+    let truncated: String = text.chars().take(char_ceiling).collect();
+    let tokens = approx_token_count(&truncated);
+    (truncated, tokens)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::openhuman::memory::tree::source_tree::types::TreeKind;
+    use chrono::Utc;
+
+    fn sample_input(id: &str, content: &str, entities: &[&str]) -> SummaryInput {
+        let ts = Utc::now();
+        SummaryInput {
+            id: id.to_string(),
+            content: content.to_string(),
+            token_count: approx_token_count(content),
+            entities: entities.iter().map(|s| s.to_string()).collect(),
+            topics: Vec::new(),
+            time_range_start: ts,
+            time_range_end: ts,
+            score: 0.5,
+        }
+    }
+
+    fn test_ctx() -> SummaryContext<'static> {
+        SummaryContext {
+            tree_id: "tree-1",
+            tree_kind: TreeKind::Source,
+            target_level: 1,
+            token_budget: 10_000,
+        }
+    }
+
+    #[tokio::test]
+    async fn concats_inputs_with_provenance_prefix() {
+        let s = InertSummariser::default();
+        let inputs = vec![
+            sample_input("a", "hello world", &[]),
+            sample_input("b", "second contribution", &[]),
+        ];
+        let out = s.summarise(&inputs, &test_ctx()).await.unwrap();
+        assert!(out.content.contains(PROVENANCE_PREFIX));
+        assert!(out.content.contains("hello world"));
+        assert!(out.content.contains("second contribution"));
+        assert_eq!(out.token_count, approx_token_count(&out.content));
+    }
+
+    #[tokio::test]
+    async fn honest_stub_emits_no_entities_or_topics() {
+        // Per design: summary-level entities/topics are LLM-derived from
+        // the summary's own synthesised content. The inert fallback does
+        // not propagate children's labels — it emits empty vecs. The
+        // Ollama summariser (future) will fill them via real NER on its
+        // own output.
+        let s = InertSummariser::default();
+        let inputs = vec![
+            sample_input("a", "x", &["entity:alice", "entity:bob"]),
+            sample_input("b", "y", &["entity:bob", "entity:carol"]),
+        ];
+        let out = s.summarise(&inputs, &test_ctx()).await.unwrap();
+        assert!(out.entities.is_empty());
+        assert!(out.topics.is_empty());
+    }
+
+    #[tokio::test]
+    async fn truncates_when_over_budget() {
+        let s = InertSummariser::default();
+        let long_text = "a".repeat(100);
+        let inputs = vec![sample_input("a", &long_text, &[])];
+        let mut ctx = test_ctx();
+        ctx.token_budget = 5; // way under — should truncate hard
+        let out = s.summarise(&inputs, &ctx).await.unwrap();
+        assert!(out.token_count <= ctx.token_budget + 1);
+        assert!(out.content.len() < long_text.len() + PROVENANCE_PREFIX.len());
+    }
+
+    #[tokio::test]
+    async fn skips_empty_contributions() {
+        let s = InertSummariser::default();
+        let inputs = vec![
+            sample_input("a", "   ", &[]),
+            sample_input("b", "kept", &[]),
+        ];
+        let out = s.summarise(&inputs, &test_ctx()).await.unwrap();
+        assert!(out.content.contains("kept"));
+        // exactly one provenance prefix should appear
+        assert_eq!(out.content.matches(PROVENANCE_PREFIX).count(), 1);
+    }
+}

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)
 }