//! Append + cascade-seal for summary trees (#709).
//!
//! `append_leaf` pushes a persisted chunk into the L0 buffer of a tree.
//! Seal gates differ by level:
//!
//! - **L0 (leaves → L1)**: seal when `token_sum >= INPUT_TOKEN_BUDGET`.
//!   Token-only gating lets small-token items (e.g. commit messages at
//!   ~20-50 tokens each) accumulate into large batches so summaries
//!   cover meaningful spans of activity.
//! - **L≥1 (summaries → next level)**: seal when `item_ids.len() >=
//!   SUMMARY_FANOUT`. Per-summary token size depends on summariser
//!   quality, so a token-based gate collapses to a 1:1:1 chain when the
//!   summariser is weak. Counting siblings keeps the tree's fan-in
//!   stable regardless.
//!
//! When a buffer seals, its items move into the new summary's
//! `child_ids`, the buffer clears, and the new summary id is queued at
//! the next level. The cascade continues upward until a buffer fails its
//! gate.
//!
//! For low-volume sources that never hit the token or sibling-count
//! thresholds, time-based sealing is handled by
//! [`flush_stale_buffers`](super::flush::flush_stale_buffers), which
//! runs on a periodic cadence and force-seals stale L0 buffers.
//!
//! 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 std::collections::BTreeSet;
use std::sync::Arc;

use anyhow::{Context, Result};
use chrono::{DateTime, Utc};
use futures::stream::{StreamExt, TryStreamExt};
use rusqlite::Transaction;

use crate::openhuman::config::Config;
use crate::openhuman::memory::util::redact::redact;
use crate::openhuman::memory_store::chunks::store::with_connection;
use crate::openhuman::memory_store::content::{
    atomic::stage_summary_with_layout,
    paths::{slugify_source_id, SummaryDiskLayout},
    wiki_git::{SummaryCommitBatch, SummaryCommitEntry},
    SummaryComposeInput,
};
use crate::openhuman::memory_store::trees::types::{
    Buffer, SummaryNode, Tree, INPUT_TOKEN_BUDGET, OUTPUT_TOKEN_BUDGET, SUMMARY_FANOUT,
};
use crate::openhuman::memory_tree::score::embed::build_write_embedder;
use crate::openhuman::memory_tree::score::extract::EntityExtractor;
use crate::openhuman::memory_tree::score::resolver::canonicalise;
use crate::openhuman::memory_tree::summarise::{
    fallback_summary, summarise, SummaryContext, SummaryInput, SummaryOutput,
};
use crate::openhuman::memory_tree::tree::factory::TreeFactory;
use crate::openhuman::memory_tree::tree::registry::new_summary_id;
use crate::openhuman::memory_tree::tree::store;

/// 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;

/// How a sealed summary node's `entities` and `topics` fields get populated.
///
/// Each tree kind has different correct semantics:
/// - **Source** trees use [`LabelStrategy::ExtractFromContent`] so the
///   summariser's freshly-synthesised text gets its own pass through an
///   extractor. Captures emergent themes that no individual leaf expressed.
/// - **Global** trees use [`LabelStrategy::UnionFromChildren`] — their
///   inputs are already-labeled source-tree summaries; union preserves
///   labels for time-based retrieval ("days that mentioned Alice")
///   without an LLM call.
/// - **Topic** trees use [`LabelStrategy::Empty`] — their scope already
///   pins the dominant theme; inheriting auxiliary entities would
///   cross-pollinate unrelated topic trees and noise the entity index.
#[derive(Clone)]
pub enum LabelStrategy {
    /// Run the extractor on the new summary's content; canonicalise the
    /// result into `entities` (canonical_ids) and `topics` (labels).
    ExtractFromContent(Arc<dyn EntityExtractor>),
    /// Dedup-merge each input's `entities` and `topics` into the parent.
    UnionFromChildren,
    /// Leave both fields empty regardless of inputs.
    Empty,
}

impl std::fmt::Debug for LabelStrategy {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::ExtractFromContent(ex) => write!(f, "ExtractFromContent({})", ex.name()),
            Self::UnionFromChildren => f.write_str("UnionFromChildren"),
            Self::Empty => f.write_str("Empty"),
        }
    }
}

/// Resolve `entities` and `topics` for a freshly-summarised node according
/// to the chosen strategy. Errors propagate from the extractor (when used).
async fn resolve_labels(
    strategy: &LabelStrategy,
    inputs: &[SummaryInput],
    summary_content: &str,
) -> Result<(Vec<String>, Vec<String>)> {
    match strategy {
        LabelStrategy::ExtractFromContent(extractor) => {
            let extracted = extractor
                .extract(summary_content)
                .await
                .context("seal-time extractor failed")?;
            let canonical = canonicalise(&extracted);
            let mut entities: Vec<String> = canonical
                .into_iter()
                .map(|c| c.canonical_id)
                .collect::<BTreeSet<_>>()
                .into_iter()
                .collect();
            entities.sort();
            let mut topics: Vec<String> = extracted
                .topics
                .into_iter()
                .map(|t| t.label)
                .collect::<BTreeSet<_>>()
                .into_iter()
                .collect();
            topics.sort();
            Ok((entities, topics))
        }
        LabelStrategy::UnionFromChildren => {
            let mut entities: BTreeSet<String> = BTreeSet::new();
            let mut topics: BTreeSet<String> = BTreeSet::new();
            for inp in inputs {
                for e in &inp.entities {
                    entities.insert(e.clone());
                }
                for t in &inp.topics {
                    topics.insert(t.clone());
                }
            }
            Ok((entities.into_iter().collect(), topics.into_iter().collect()))
        }
        LabelStrategy::Empty => Ok((Vec::new(), Vec::new())),
    }
}

/// 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.
///
/// `strategy` controls how each sealed summary's `entities` and `topics`
/// are populated — see [`LabelStrategy`].
pub async fn append_leaf(
    config: &Config,
    tree: &Tree,
    leaf: &LeafRef,
    strategy: &LabelStrategy,
) -> Result<Vec<String>> {
    log::debug!(
        "[tree::bucket_seal] append_leaf tree_id={} leaf_id={} tokens={} strategy={:?}",
        tree.id,
        leaf.chunk_id,
        leaf.token_count,
        strategy
    );

    // 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, strategy).await
}

/// Queue-oriented variant of [`append_leaf`].
///
/// This only appends the leaf to the L0 buffer and returns whether the
/// caller should enqueue a follow-up seal job for level 0.
pub fn append_leaf_deferred(config: &Config, tree: &Tree, leaf: &LeafRef) -> Result<bool> {
    append_to_buffer(
        config,
        &tree.id,
        0,
        &leaf.chunk_id,
        leaf.token_count as i64,
        leaf.timestamp,
    )?;
    let buf = store::get_buffer(config, &tree.id, 0)?;
    Ok(should_seal(&buf))
}

/// Transactionally append a single item to `(tree_id, level)`'s buffer.
pub 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!(
                "[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,
    strategy: &LabelStrategy,
) -> Result<Vec<String>> {
    cascade_all_from(config, tree, 0, None, strategy).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.
///
/// `strategy` is forwarded to every sealed level — same semantics as
/// [`append_leaf`].
pub async fn cascade_all_from(
    config: &Config,
    tree: &Tree,
    start_level: u32,
    force_now: Option<DateTime<Utc>>,
    strategy: &LabelStrategy,
) -> 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!(
                "[tree::bucket_seal] cascade done tree_id={} stop_level={} token_sum={}",
                tree.id,
                level,
                buf.token_sum
            );
            break;
        }
        if buf.is_empty() {
            log::debug!(
                "[tree::bucket_seal] cascade hit empty buffer tree_id={} level={} — stopping",
                tree.id,
                level
            );
            break;
        }

        // Sync cascade — drives the level walk itself; doesn't need the
        // queue follow-ups (we'll hit `seal_one_level` again next iter).
        let summary_id = seal_one_level(config, tree, &buf, strategy, false).await?;
        sealed_ids.push(summary_id);
        level += 1;
    }

    Ok(sealed_ids)
}

/// Level-aware seal gate.
///
/// L0 buffers gate on `token_sum >= INPUT_TOKEN_BUDGET` **only** — no
/// sibling-count fallback. Token-only gating is deliberate (see the
/// module header): small-token items (commit messages, short emails)
/// accumulate into large batches so each L1 summary covers a
/// meaningful span instead of 10 tiny items. Small buffers that never
/// reach the budget are sealed by time instead:
///
/// Time-based sealing for low-volume sources is handled separately
/// by `flush_stale_buffers` (see [`crate::openhuman::memory_tree::
/// tree::flush::flush_stale_buffers`]), which filters buffers
/// by `oldest_at` before calling the cascade (the 3-hourly
/// `memory_queue` scheduler drives it). Keeping the time gate
/// out of `should_seal` avoids prematurely sealing buffers during
/// normal `append_leaf` calls when test/restored data carries older
/// timestamps.
///
/// L≥1 buffers gate on sibling count alone so the tree's
/// fan-in is independent of per-summary token size — without this,
/// summarisers that emit at the full token budget (e.g. the inert
/// fallback) collapse the cascade into a 1:1:1 chain instead of a
/// real tree.
pub(crate) fn should_seal(buf: &Buffer) -> bool {
    if buf.is_empty() {
        return false;
    }
    if buf.level == 0 {
        buf.token_sum >= INPUT_TOKEN_BUDGET as i64
    } else {
        (buf.item_ids.len() as u32) >= SUMMARY_FANOUT
    }
}

/// Seal `buf` at `level` into one summary at `level + 1`. Returns the new
/// summary id.
///
/// `strategy` decides how `entities` and `topics` get populated on the new
/// summary node — see [`LabelStrategy`].
///
/// When `enqueue_follow_ups` is `true`, the function additionally inserts
/// follow-up job rows **inside the same transaction** that commits the
/// seal:
/// - `seal { tree_id, level: parent_level }` if the parent buffer's gate
///   is now met (parent-cascade enqueue)
/// - `topic_route { NodeRef::Summary { summary_id } }` for source trees
///   (so summary-level entities feed the topic-tree spawn pipeline)
///
/// Atomic enqueue eliminates the crash window where a seal commits but
/// the post-commit follow-up enqueues are silently lost on a worker
/// crash. The async-pipeline handler (`handle_seal`) passes `true`. The
/// synchronous in-process cascade caller ([`cascade_all_from`]) passes
/// `false` because it drives the cascade itself and topic_route isn't
/// part of the test/flush sync path.
pub(crate) async fn seal_one_level(
    config: &Config,
    tree: &Tree,
    buf: &Buffer,
    strategy: &LabelStrategy,
    enqueue_follow_ups: bool,
) -> 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!(
            "[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);

    crate::core::event_bus::publish_global(
        crate::core::event_bus::DomainEvent::MemoryTreeBuildProgress {
            phase: "seal".to_string(),
            step: "summarising".to_string(),
            tree_scope: Some(tree.scope.clone()),
            level: Some(level),
            item_count: Some(inputs.len() as u32),
            detail: Some(format!("{} inputs → L{}", inputs.len(), level + 1)),
        },
    );

    let ctx = SummaryContext {
        tree_id: &tree.id,
        tree_kind: tree.kind,
        target_level,
        token_budget: OUTPUT_TOKEN_BUDGET,
    };
    // #13021: treat a blank summary (LLM returned only whitespace, so
    // `summarise()` succeeded with `content = ""`) the same as a hard error —
    // fall back to the deterministic `fallback_summary` so we never persist a
    // parent node with `content = ""` / `token_count = 0`. Without this, the
    // child text is lost: the level buffer is cleared and the parent has no
    // recoverable content for the next rollup or for retrieval.
    let output = match summarise(config, &inputs, &ctx).await {
        Ok(o) if !o.content.trim().is_empty() => o,
        Ok(_) => {
            log::warn!(
                "[memory_tree::seal] summarise returned blank for tree_id={} level={} — using fallback (#13021)",
                ctx.tree_id, ctx.target_level,
            );
            fallback_summary(&inputs, ctx.token_budget)
        }
        Err(e) => {
            log::warn!(
                "[memory_tree::seal] summarise failed for tree_id={} level={}: {e:#} — using fallback",
                ctx.tree_id, ctx.target_level,
            );
            fallback_summary(&inputs, ctx.token_budget)
        }
    };

    // Resolve labels (entities/topics) for the new summary node according
    // to the chosen strategy. Done before the write tx so an extractor
    // failure aborts the seal cleanly — same shape as the embedder guard
    // below.
    let (node_entities, node_topics) = resolve_labels(strategy, &inputs, &output.content).await?;

    // Phase 4 (#710): embed the summary BEFORE opening the write tx so an
    // embedder failure aborts the seal cleanly — nothing is persisted,
    // the buffer stays intact, and a retry re-embeds from scratch. The
    // tx below would otherwise commit a summary with no embedding,
    // polluting retrieval's semantic rerank.
    //
    // Embedder context-window guard: `nomic-embed-text-v1.5` accepts
    // up to 8192 tokens of input. Summary content is bounded by
    // `ctx.token_budget = OUTPUT_TOKEN_BUDGET = 5_000` which fits, but
    // we still truncate the input passed to `embed()` to leave
    // headroom for tokenizer drift (the persisted summary content
    // stays full; only the embedding's "view" of it is clamped).
    crate::core::event_bus::publish_global(
        crate::core::event_bus::DomainEvent::MemoryTreeBuildProgress {
            phase: "seal".to_string(),
            step: "embedding".to_string(),
            tree_scope: Some(tree.scope.clone()),
            level: Some(level),
            item_count: None,
            detail: None,
        },
    );

    // Conservative cap. Slack-style chat content (URLs, mentions,
    // emoji) tokenizes 2-4× higher than the 4-chars/token heuristic.
    // 1000 approx-tokens (~4000 chars) is comfortably under 8192
    // even at 4× tokenizer ratio.
    let embed_input = truncate_for_embed(&output.content, 1_000);
    log::info!(
        "[tree::bucket_seal] embed input: original_chars={} truncated_chars={}",
        output.content.len(),
        embed_input.len()
    );
    // #002 (FR-002): skip embedding when no usable provider is configured
    // (build_write_embedder returns None) rather than writing a fake all-zero
    // vector. The summary is sealed embedding-less (re-embeddable later) and
    // the semantic-recall degraded flag is already set with a typed cause.
    //
    // #13021: also skip when `embed_input.trim().is_empty()`. `summarise()`
    // returns `SummaryOutput::default()` (content = "") when every input is
    // whitespace, and `fallback_summary` joins zero parts the same way. An
    // empty `embed_input` is guaranteed to 400 from the upstream embedding
    // API — short-circuit to the same "embedding-less, re-embeddable later"
    // path as the no-provider case.
    let embedding: Option<Vec<f32>> = if embed_input.trim().is_empty() {
        log::warn!(
            "[tree::bucket_seal] empty summary content for tree_id={} level={}→{} \
                 — sealing without embedding (#13021)",
            tree.id,
            level,
            target_level
        );
        None
    } else {
        match build_write_embedder(config).context("build embedder during seal")? {
            None => {
                log::warn!(
                    "[tree::bucket_seal] embeddings unavailable for tree_id={} level={}→{} \
                     — sealing summary without embedding (semantic recall degraded)",
                    tree.id,
                    level,
                    target_level
                );
                None
            }
            Some(embedder) => {
                let v = match embedder.embed(&embed_input).await {
                    Ok(v) => v,
                    Err(e) => {
                        // #002: classify so the seal job fails fast on
                        // unrecoverable embed causes (budget/auth/dim) with a
                        // typed reason instead of retrying; original chain
                        // preserved as context.
                        let failure =
                            crate::openhuman::memory_tree::health::classify_embed_error(&e);
                        return Err(anyhow::Error::new(failure).context(format!(
                            "embed summary during seal tree_id={} level={}: {e:#}",
                            tree.id, level
                        )));
                    }
                };
                // Dimension guard: reject wrong-dimensionality vectors before
                // they reach the store — same contract as handle_extract's
                // pack_checked. Without this a provider returning the wrong
                // shape slips into the summary sidecar silently.
                crate::openhuman::memory_tree::score::embed::pack_checked(&v).context(format!(
                    "seal embed dim check tree_id={} level={}",
                    tree.id, level
                ))?;
                log::debug!(
                "[tree::bucket_seal] embedded summary tree_id={} level={}→{} bytes={} provider={}",
                tree.id,
                level,
                target_level,
                output.content.len(),
                embedder.name()
            );
                crate::openhuman::memory_tree::health::clear_semantic_recall_degraded();
                Some(v)
            }
        }
    };

    // 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(),
        // `seal_one_level` runs for source AND topic trees (handle_seal,
        // cascade_all_from, flush). Hardcoding Source here would write
        // topic-tree summaries with tree_kind='source' in
        // mem_tree_summaries, breaking any query filtering on tree_kind.
        tree_kind: tree.kind,
        level: target_level,
        parent_id: None,
        child_ids: buf.item_ids.clone(),
        content: output.content,
        token_count: output.token_count,
        entities: node_entities,
        topics: node_topics,
        time_range_start,
        time_range_end,
        score,
        sealed_at: now,
        deleted: false,
        embedding,
        // Generic seal path (chat/email source trees + the cross-document
        // merge tier) is document-agnostic. The per-document subtree seal
        // (Notion) sets these via its own seal entrypoint in Task #2.
        doc_id: None,
        version_ms: None,
    };

    crate::core::event_bus::publish_global(
        crate::core::event_bus::DomainEvent::MemoryTreeBuildProgress {
            phase: "seal".to_string(),
            step: "persisting".to_string(),
            tree_scope: Some(tree.scope.clone()),
            level: Some(target_level),
            item_count: None,
            detail: Some(format!(
                "summary {} ({} tokens)",
                &summary_id[..summary_id.len().min(16)],
                output.token_count
            )),
        },
    );

    // Phase MD-content: stage the summary .md file BEFORE opening the write
    // tx. A staging failure aborts the seal cleanly — nothing is persisted
    // and the buffer stays intact for retry.
    //
    let tree_factory = TreeFactory::from_tree(tree);
    let summary_tree_kind = tree_factory.summary_tree_kind();
    let scope_slug = tree_factory.scope_slug();
    // For L1 seals (children are chunks), point each child wikilink at
    // the raw archive file the chunk's body lives in — the email
    // chunk-store path `email/<scope>/<chunk_id>.md` no longer
    // exists, so `[[<chunk_id>]]` would be an unresolved Obsidian
    // link. We emit the relative path under content_root (with `.md`
    // stripped) so the wikilink resolves unambiguously even outside
    // Obsidian's unique-basename heuristic — e.g.
    // `[[raw/gmail-stevent95-at-gmail-dot-com/<ts_ms>_<msg_id>]]`.
    // L≥2 children are summary ids whose default `sanitize_filename`
    // resolves to existing `wiki/summaries/...md` files — leave
    // overrides unset there.
    let child_basename_overrides: Option<Vec<Option<String>>> = if node.level == 1 {
        let overrides: Vec<Option<String>> = node
            .child_ids
            .iter()
            .map(|chunk_id| {
                // Surface lookup failures explicitly — silently
                // falling back to `[[<chunk_hash>]]` would commit an
                // unresolved Obsidian wikilink without any signal.
                // We still yield `None` (so `compose_summary_md`
                // takes the sanitised-id fallback) but a warn log
                // makes the SQL error visible for diagnosis.
                match crate::openhuman::memory_store::chunks::store::get_chunk_raw_refs(
                    config, chunk_id,
                ) {
                    Ok(Some(refs)) if !refs.is_empty() => {
                        // RawRef::path is a forward-slash relative path
                        // under content_root, e.g.
                        // "raw/gmail-…/1700000_msg-id.md". Strip `.md`
                        // for Obsidian's extension-less wikilink
                        // convention.
                        let r = refs.into_iter().next().expect("non-empty");
                        Some(r.path.strip_suffix(".md").unwrap_or(&r.path).to_string())
                    }
                    Ok(_) => {
                        // No raw_refs persisted for this chunk — most
                        // commonly slack chunks (we only stage raw
                        // archive files for gmail today). The wikilink
                        // falls back to `sanitize_filename(chunk_id)`,
                        // which produces a deliberately-unresolved
                        // Obsidian link. Log so the silent-degradation
                        // path stays visible during diagnosis.
                        log::debug!(
                            "[tree::bucket_seal] no raw_refs for chunk_id={chunk_id} \
                             — wikilink will fall back to sanitised chunk id"
                        );
                        None
                    }
                    Err(e) => {
                        log::warn!(
                            "[tree::bucket_seal] get_chunk_raw_refs failed \
                             chunk_id={chunk_id} err={e:#} — falling back to \
                             chunk_id-based wikilink"
                        );
                        None
                    }
                }
            })
            .collect();
        Some(overrides)
    } else {
        None
    };
    let compose_input = SummaryComposeInput {
        summary_id: &node.id,
        tree_kind: summary_tree_kind,
        tree_id: &node.tree_id,
        tree_scope: &tree.scope,
        level: node.level,
        child_ids: &node.child_ids,
        child_basenames: child_basename_overrides.as_deref(),
        child_count: node.child_ids.len(),
        time_range_start: node.time_range_start,
        time_range_end: node.time_range_end,
        sealed_at: node.sealed_at,
        body: &node.content,
    };
    // Stage the summary .md file and propagate any error — a staging failure
    // aborts the seal entirely so the database never commits a row with
    // content_path = NULL. The buffer stays unsealed and the job-retry path
    // will re-attempt the file write on next execution.
    let content_root = config.memory_tree_content_root();
    // Drop the bundled `.obsidian/` defaults (graph + types) so a user
    // opening the vault gets the intended graph-view colour mapping
    // without manual configuration. Best-effort and idempotent — never
    // overwrites an existing file.
    if let Err(err) =
        crate::openhuman::memory_store::content::obsidian::ensure_obsidian_defaults(&content_root)
    {
        log::warn!(
            "[tree::bucket_seal] ensure_obsidian_defaults failed: {err:#} — \
             continuing seal without vault defaults"
        );
    }
    // Merge-tier nodes (document source trees, level ≥ MERGE_LEVEL_BASE) land
    // under `source-<scope>/merge/`; everything else (chat/email + the
    // per-doc subtree is sealed via seal_explicit_children, not here) uses the
    // flat layout.
    let layout = if node.level >= MERGE_LEVEL_BASE {
        SummaryDiskLayout::Merge
    } else {
        SummaryDiskLayout::Standard
    };
    let staged = stage_summary_with_layout(&content_root, &compose_input, &scope_slug, layout)
        .with_context(|| {
            format!(
                "stage_summary failed for {}; seal aborted, buffer stays unsealed for retry",
                node.id
            )
        })?;
    log::debug!(
        "[tree::bucket_seal] staged summary {} → {}",
        node.id,
        staged.content_path
    );
    let summary_commit = summary_seal_batch(
        tree,
        "bucket_seal",
        std::slice::from_ref(&node),
        std::slice::from_ref(&staged.content_path),
    )
    .with_context(|| {
        format!(
            "build summary git batch failed for {}; seal aborted, buffer stays unsealed for retry",
            node.id
        )
    })?;

    // Single write transaction: insert summary, clear this buffer, append
    // summary id to parent buffer, bump tree max_level/root if needed,
    // and (when `enqueue_follow_ups`) atomically enqueue parent-seal +
    // topic_route follow-ups so they can never desync from the commit.
    // 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,
            Some(&staged),
            &crate::openhuman::memory_store::chunks::store::tree_active_signature(config),
        )?;
        // 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 when entities is empty (the current summarise()
        // always emits empty — entity extraction is the learning domain's job);
        // becomes active once the summariser or a post-seal extractor emits 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)?;

        // Atomic follow-up enqueues. Done INSIDE this tx — if the commit
        // rolls back, the queue rows go with it; if it succeeds, the
        // rows are durably visible to the worker pool. Eliminates the
        // crash window where the seal commits but post-commit enqueues
        // are lost.
        if enqueue_follow_ups {
            // Parent-cascade: if the new summary made the parent buffer
            // cross its gate, enqueue the next level's seal. Dedupe key
            // `seal:{tree_id}:{parent_level}` prevents duplicates if a
            // parallel path already queued it.
            if should_seal(&parent) {
                use crate::openhuman::memory_queue::store::enqueue_tx as enqueue_job_tx;
                use crate::openhuman::memory_queue::types::{NewJob, SealPayload};
                let parent_seal = SealPayload {
                    tree_id: tree_id.clone(),
                    level: target_level_for_closure,
                    force_now_ms: None,
                };
                enqueue_job_tx(&tx, &NewJob::seal(&parent_seal)?)?;
            }
            // (Topic-tree routing removed: the topic/global trees were
            // deleted — source trees plus the entity index are the
            // substrate, so a source seal no longer fans out anywhere.)
        }

        // 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(())
    })?;

    commit_summary_seal(config, &summary_commit).with_context(|| {
        format!(
            "commit_summary_seal failed for {} after DB seal commit",
            summary_id
        )
    })?;

    log::info!(
        "[tree::bucket_seal] sealed tree_id={} level={}→{} summary_id={} children={}",
        tree.id,
        level,
        target_level,
        summary_id,
        buf.item_ids.len()
    );

    Ok(summary_id)
}

/// Clamp `text` to roughly `max_tokens` tokens before passing to the
/// embedder. Uses the same ~4 chars/token heuristic as
/// `approx_token_count`. Embedders have hard input-size limits (e.g.
/// `nomic-embed-text-v1.5` = 8192 tokens) and an overshoot returns
/// HTTP 500 from Ollama rather than auto-truncating, which would
/// abort the seal transaction.
fn truncate_for_embed(text: &str, max_tokens: u32) -> String {
    let approx = crate::openhuman::memory_store::chunks::types::approx_token_count(text);
    if approx <= max_tokens {
        return text.to_string();
    }
    let char_ceiling = (max_tokens as usize).saturating_mul(4);
    text.chars().take(char_ceiling).collect()
}

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

fn summary_seal_batch(
    tree: &Tree,
    reason: &str,
    nodes: &[SummaryNode],
    content_paths: &[String],
) -> Result<SummaryCommitBatch> {
    if nodes.is_empty() {
        return Ok(SummaryCommitBatch {
            reason: reason.to_string(),
            tree_id: tree.id.clone(),
            tree_scope: tree.scope.clone(),
            entries: Vec::new(),
        });
    }
    if nodes.len() != content_paths.len() {
        anyhow::bail!(
            "[tree::bucket_seal] commit_summary_seal: node/path length mismatch nodes={} paths={}",
            nodes.len(),
            content_paths.len()
        );
    }

    let entries = nodes
        .iter()
        .zip(content_paths.iter())
        .map(|(node, content_path)| SummaryCommitEntry {
            summary_id: node.id.clone(),
            content_path: content_path.clone(),
            level: node.level,
            child_count: node.child_ids.len(),
            token_count: node.token_count,
            time_range_start: node.time_range_start,
            time_range_end: node.time_range_end,
        })
        .collect();

    Ok(SummaryCommitBatch {
        reason: reason.to_string(),
        tree_id: tree.id.clone(),
        tree_scope: tree.scope.clone(),
        entries,
    })
}

fn commit_summary_seal(config: &Config, batch: &SummaryCommitBatch) -> Result<()> {
    crate::openhuman::memory_store::content::wiki_git::commit_summaries(
        &config.memory_tree_content_root(),
        batch,
    )
}

/// 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_store::chunks::store::get_chunk;
    use crate::openhuman::memory_store::content::read as content_read;
    use crate::openhuman::memory_tree::score::store::{get_score, list_entity_ids_for_node};

    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!(
                    "[tree::bucket_seal] hydrate_leaf_inputs: missing chunk {id} — skipping"
                );
                continue;
            }
        };
        let score_value = get_score(config, id)?.map(|row| row.total).unwrap_or(0.0);
        // Pull canonical entity ids from the inverted index — that's the
        // authoritative source for "what entities are attached to this
        // chunk." Topics live on the chunk's metadata tags.
        // [`LabelStrategy::UnionFromChildren`] reads these fields off
        // each `SummaryInput` to roll labels up the tree.
        let entities = list_entity_ids_for_node(config, id).unwrap_or_default();
        // Read the full body from disk — the `content` column in SQLite holds
        // a ≤500-char preview after the MD-on-disk migration. The summariser
        // must receive the complete chunk text so the seal output is not a
        // summary of previews.
        //
        // For pre-MD-migration chunks (no content_path recorded) this call
        // returns Err; callers that want to handle legacy rows should check
        // content_path presence before calling hydrate_inputs.
        let body = content_read::read_chunk_body(config, id).with_context(|| {
            format!("[tree::bucket_seal] hydrate_leaf_inputs: read body for chunk {id}")
        })?;
        out.push(SummaryInput {
            id: chunk.id.clone(),
            content: body,
            token_count: chunk.token_count,
            entities,
            topics: chunk.metadata.tags.clone(),
            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>> {
    use crate::openhuman::memory_store::content::read as content_read;
    use crate::openhuman::memory_store::trees::store::get_summaries_batch;

    // One batched `SELECT … WHERE id IN (?,?,…)` instead of N per-id
    // `get_summary` round-trips. Body reads stay per-id because each
    // summary's full markdown lives in its own on-disk file — batching
    // there would mean concurrent file opens, not a single round-trip.
    // Walking the caller's `summary_ids` slice (not the map) preserves
    // input order, matching the previous per-id loop's semantics; the
    // map lookup keys by id so the order of `HashMap`'s iteration is
    // irrelevant.
    let node_by_id = get_summaries_batch(config, summary_ids)?;

    let mut out: Vec<SummaryInput> = Vec::with_capacity(summary_ids.len());
    for id in summary_ids {
        let Some(node) = node_by_id.get(id) else {
            log::warn!(
                "[tree::bucket_seal] hydrate_summary_inputs: missing summary {id} — skipping"
            );
            continue;
        };
        // Read the full body from disk — `node.content` is a ≤500-char preview
        // after the MD-on-disk migration. Higher-level seals (L2+) summarise
        // over L1 summary content and need the full text, not a preview.
        let body = content_read::read_summary_body(config, id).with_context(|| {
            format!("[tree::bucket_seal] hydrate_summary_inputs: read body for summary {id}")
        })?;
        out.push(SummaryInput {
            id: node.id.clone(),
            content: body,
            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)
}

// ── Document-aware sealing (Notion etc.) ────────────────────────────────
//
// Document source trees keep ONE physical `mem_tree_trees` row per
// connection (e.g. `notion:{connection_id}`), but inside it each document
// rolls up to its own **doc-root** summary, and those doc-roots merge into
// the connection root. To get that shape without re-keying the shared
// `(tree, level)` buffer — the exact path chat/email seal through — the
// per-document subtree is built as an **isolated side-cascade**
// ([`seal_document_subtree`]) that never touches a shared buffer or the
// tree root. Only the cross-document **merge tier** uses the shared buffer
// + the existing [`cascade_all_from`] engine, starting at
// [`MERGE_LEVEL_BASE`].
//
// Versioning is forward-only: editing a Notion page calls
// [`seal_document_subtree`] again with a higher `version_ms`, producing a
// *new* doc-root that is appended to the merge buffer alongside the old
// one. Nothing is rewritten or tombstoned; retrieval keeps `max(version_ms)`
// per `doc_id` at read time (see the retrieval layer).

/// Level offset where the cross-document merge tier lives inside a document
/// source tree.
///
/// Per-document subtrees occupy small levels (1, 2, …) and are built by
/// [`seal_document_subtree`] as a side-cascade that never enters a shared
/// `(tree, level)` buffer. The merge tier — which summarises *across*
/// documents — uses the shared buffer and the existing cascade engine,
/// starting here. The wide gap guarantees per-doc nodes (small levels) and
/// merge nodes (≥ `MERGE_LEVEL_BASE`) can never collide on `(tree, level)`,
/// and keeps `Tree.root_id` / `max_level` pointing at a merge node.
pub const MERGE_LEVEL_BASE: u32 = 1_000;

/// Hard cap on how many children one per-document summary fans in, so a
/// single huge document can't produce a doc-root with thousands of direct
/// children. Independent of [`SUMMARY_FANOUT`] (which gates the merge tier).
const DOC_SUBTREE_MAX_FANIN: usize = 32;

/// Bound on how many sibling batches within a single document-subtree level
/// are sealed concurrently. Each [`seal_explicit_children`] call is one
/// independent summarise + embed round-trip pair, so overlapping them with a
/// small pool collapses the serial `N × (LLM + embed RTT)` wall-time toward
/// `N/CONCURRENCY × RTT` on the ingest path. Kept low: the SQLite write tx
/// serialises on a single writer anyway, and networked summarise/embed
/// providers (Voyage, cloud LLMs) rate-limit — so a large pool buys nothing
/// and risks 429s. `buffered` (ordered) preserves the serial cascade's batch
/// order, so the resulting tree shape is identical.
const DOC_SUBTREE_SEAL_CONCURRENCY: usize = 8;

/// Build (or re-build, for a new version) one document's subtree and merge
/// its doc-root into the connection tree.
///
/// `doc_id` is the document identity (the chunk `source_id`, e.g.
/// `notion:{conn}:{page_id}`); `version_ms` is the document version
/// (`last_edited_time` epoch-ms). `chunk_ids` are this version's leaf chunk
/// ids, already persisted in `mem_tree_chunks`.
///
/// Steps:
/// 1. Cascade `chunk_ids` upward (token-budget batches at L0, count batches
///    above) until a **single doc-root** remains — force-sealed even for a
///    one-chunk document so it always surfaces as a doc-root, never loose
///    leaves. Every node is tagged `(doc_id, version_ms)`.
/// 2. Append the doc-root to the connection tree's merge buffer at
///    [`MERGE_LEVEL_BASE`] and run the existing cascade so it folds into the
///    connection root once `SUMMARY_FANOUT` doc-roots accumulate.
///
/// Returns the doc-root summary id. Idempotent re-runs for the *same*
/// `(doc_id, version_ms, chunk_ids)` produce a new doc-root (new ids); the
/// caller (Notion sync) only invokes this when a new revision is admitted.
pub async fn seal_document_subtree(
    config: &Config,
    tree: &Tree,
    doc_id: &str,
    version_ms: Option<i64>,
    chunk_ids: &[String],
    strategy: &LabelStrategy,
) -> Result<String> {
    if chunk_ids.is_empty() {
        anyhow::bail!(
            "[tree::bucket_seal] seal_document_subtree: empty chunk set tree_id={} doc_id_hash={}",
            tree.id,
            redact(doc_id)
        );
    }
    log::debug!(
        "[tree::bucket_seal] seal_document_subtree tree_id={} doc_id_hash={} version_ms={:?} chunks={}",
        tree.id,
        redact(doc_id),
        version_ms,
        chunk_ids.len()
    );

    // 1. Per-document side-cascade to a single doc-root.
    let mut current_level: u32 = 0;
    let mut current_ids: Vec<String> = chunk_ids.to_vec();
    let mut doc_root: Option<SummaryNode> = None;

    loop {
        let batches = if current_level == 0 {
            batch_leaves_by_token_budget(config, &current_ids)?
        } else {
            batch_by_count(&current_ids, DOC_SUBTREE_MAX_FANIN)
        };

        // Within one level the batches are disjoint child sets, and
        // `seal_explicit_children` touches no shared `(tree, level)` buffer
        // and never advances the tree root (see its docstring) — so the
        // sibling seals are independent and their summarise + embed
        // round-trips can overlap. Run them with bounded concurrency.
        //
        // `buffered` (ordered), NOT `buffer_unordered`: the next level groups
        // `next_ids` positionally via `batch_by_count`, so the output order
        // must match the serial cascade to keep the tree shape identical.
        // `try_collect` short-circuits on the first error, preserving the
        // old `?`-on-first-failure behaviour. Levels stay sequential — level
        // N+1 consumes level N's output — so only siblings overlap, not the
        // cross-level fan-in. The blocking SQLite write inside each seal
        // still serialises (single writer), which is fine: the win is the
        // overlapped network RTTs, not the DB writes.
        // Build the per-batch futures eagerly into a `Vec` (they don't run until
        // polled) rather than lazily inside a `stream::iter(...).map(closure)`:
        // the lazy form makes rustc try to infer a higher-ranked `FnOnce` over
        // the borrowed `batch` lifetime, which fails ("implementation of
        // `FnOnce` is not general enough") once this async fn is spawned. Eager
        // collection ties each future to this stack frame's concrete lifetime.
        if batches.len() > 1 {
            log::debug!(
                "[tree::bucket_seal] doc-subtree concurrent seal tree_id={} doc_id_hash={} level={} batches={} concurrency={}",
                tree.id,
                redact(doc_id),
                current_level,
                batches.len(),
                DOC_SUBTREE_SEAL_CONCURRENCY
            );
        }
        let batch_futures: Vec<_> = batches
            .iter()
            .map(|batch| {
                seal_explicit_children(
                    config,
                    tree,
                    current_level,
                    batch,
                    Some(doc_id),
                    version_ms,
                    strategy,
                )
            })
            .collect();
        let nodes: Vec<SummaryNode> = futures::stream::iter(batch_futures)
            .buffered(DOC_SUBTREE_SEAL_CONCURRENCY)
            .try_collect()
            .await?;

        let next_ids: Vec<String> = nodes.iter().map(|n| n.id.clone()).collect();
        // The terminal level produces exactly one node (the loop breaks once
        // `current_ids.len() <= 1`); preserving batch order means the last
        // node is the doc-root, identical to the serial assignment.
        doc_root = nodes.into_iter().next_back();

        current_level += 1;
        current_ids = next_ids;
        if current_ids.len() <= 1 {
            break;
        }
    }

    let doc_root = doc_root.ok_or_else(|| {
        anyhow::anyhow!(
            "[tree::bucket_seal] seal_document_subtree produced no doc-root tree_id={} doc_id_hash={}",
            tree.id,
            redact(doc_id)
        )
    })?;
    log::debug!(
        "[tree::bucket_seal] doc-root sealed tree_id={} doc_id_hash={} root_id={} level={}",
        tree.id,
        redact(doc_id),
        doc_root.id,
        doc_root.level
    );

    // 2. Feed the doc-root into the cross-document merge tier and cascade
    //    using the untouched shared engine.
    append_to_buffer(
        config,
        &tree.id,
        MERGE_LEVEL_BASE,
        &doc_root.id,
        doc_root.token_count as i64,
        doc_root.time_range_start,
    )?;
    let merge_sealed = cascade_all_from(config, tree, MERGE_LEVEL_BASE, None, strategy).await?;
    log::debug!(
        "[tree::bucket_seal] merge cascade tree_id={} doc_id_hash={} merge_sealed={}",
        tree.id,
        redact(doc_id),
        merge_sealed.len()
    );

    Ok(doc_root.id)
}

/// Greedily batch leaf chunk ids so each batch stays under
/// [`INPUT_TOKEN_BUDGET`] (and at most [`DOC_SUBTREE_MAX_FANIN`] children).
/// A single oversized chunk forms its own batch.
fn batch_leaves_by_token_budget(config: &Config, chunk_ids: &[String]) -> Result<Vec<Vec<String>>> {
    use crate::openhuman::memory_store::chunks::store::get_chunk;

    let mut batches: Vec<Vec<String>> = Vec::new();
    let mut current: Vec<String> = Vec::new();
    let mut token_sum: i64 = 0;

    for id in chunk_ids {
        let tokens = match get_chunk(config, id)? {
            Some(c) => c.token_count as i64,
            None => {
                log::warn!(
                    "[tree::bucket_seal] batch_leaves_by_token_budget: missing chunk {id} — skipping"
                );
                continue;
            }
        };
        let would_exceed = token_sum + tokens > INPUT_TOKEN_BUDGET as i64
            || current.len() >= DOC_SUBTREE_MAX_FANIN;
        if would_exceed && !current.is_empty() {
            batches.push(std::mem::take(&mut current));
            token_sum = 0;
        }
        current.push(id.clone());
        token_sum += tokens;
    }
    if !current.is_empty() {
        batches.push(current);
    }
    if batches.is_empty() {
        // All chunks were missing — surface as one empty batch caller rejects.
        anyhow::bail!("[tree::bucket_seal] batch_leaves_by_token_budget: no resolvable chunks");
    }
    Ok(batches)
}

/// Split ids into fixed-size batches of at most `max` (used above L0 in the
/// per-document cascade).
fn batch_by_count(ids: &[String], max: usize) -> Vec<Vec<String>> {
    ids.chunks(max.max(1)).map(|c| c.to_vec()).collect()
}

/// Seal an **explicit** set of child ids into one summary at `level + 1`,
/// tagging it with `doc_id` / `version_ms`. Unlike [`seal_one_level`] this
/// does NOT touch any shared `(tree, level)` buffer and does NOT advance the
/// tree root/max_level — it is the per-document side-cascade primitive. It
/// reuses the same hydrate → summarise → label → embed → stage → persist
/// pipeline so doc-subtree summaries are indistinguishable from regular
/// summaries except for their `doc_id` / `version_ms` tags.
async fn seal_explicit_children(
    config: &Config,
    tree: &Tree,
    level: u32,
    child_ids: &[String],
    doc_id: Option<&str>,
    version_ms: Option<i64>,
    strategy: &LabelStrategy,
) -> Result<SummaryNode> {
    let target_level = level + 1;
    let inputs = hydrate_inputs(config, level, child_ids)?;
    if inputs.is_empty() {
        anyhow::bail!(
            "[tree::bucket_seal] seal_explicit_children: empty inputs tree_id={} level={}",
            tree.id,
            level
        );
    }

    let time_range_start = inputs
        .iter()
        .map(|i| i.time_range_start)
        .min()
        .unwrap_or_else(Utc::now);
    let time_range_end = inputs
        .iter()
        .map(|i| i.time_range_end)
        .max()
        .unwrap_or_else(Utc::now);
    let score = inputs
        .iter()
        .map(|i| i.score)
        .fold(f32::NEG_INFINITY, f32::max)
        .max(0.0);

    let ctx = SummaryContext {
        tree_id: &tree.id,
        tree_kind: tree.kind,
        target_level,
        token_budget: OUTPUT_TOKEN_BUDGET,
    };
    // Single-input passthrough: if a doc rolls up from exactly one node that
    // already fits the summary budget (the common case — a Notion page that is
    // a single chunk), there is nothing to summarise. Emit the input verbatim
    // as the doc-root content and SKIP the LLM call entirely. The doc-root is
    // still a real summary node (so versioning, the merge tier, and the
    // read-time latest-version filter all keep working uniformly) — it just
    // isn't a redundant paraphrase of one chunk, and costs no inference.
    // A single oversized input still goes through the summariser (it genuinely
    // needs compression).
    let output = if inputs.len() == 1 && inputs[0].token_count <= OUTPUT_TOKEN_BUDGET {
        log::debug!(
            "[tree::bucket_seal] doc-subtree passthrough (1 input, no LLM) tree_id={} doc_id_hash={} level={}",
            tree.id,
            doc_id.map(redact).unwrap_or_default(),
            level
        );
        let only = &inputs[0];
        SummaryOutput {
            content: only.content.clone(),
            token_count: only.token_count,
            entities: Vec::new(),
            topics: Vec::new(),
            input_tokens: 0,
            output_tokens: 0,
            charged_amount_usd: None,
        }
    } else {
        // #13021: blank summary (whitespace-only LLM output) → fallback. See
        // the matching branch in `seal_one_level` for why we treat blank
        // content as a hard fail rather than persisting `content = ""`.
        match summarise(config, &inputs, &ctx).await {
            Ok(o) if !o.content.trim().is_empty() => o,
            Ok(_) => {
                log::warn!(
                    "[tree::bucket_seal] doc-subtree summarise returned blank tree_id={} doc_id_hash={} level={} — fallback (#13021)",
                    tree.id, doc_id.map(redact).unwrap_or_default(), level,
                );
                fallback_summary(&inputs, ctx.token_budget)
            }
            Err(e) => {
                log::warn!(
                    "[tree::bucket_seal] doc-subtree summarise failed tree_id={} doc_id_hash={} level={}: {e:#} — fallback",
                    tree.id, doc_id.map(redact).unwrap_or_default(), level,
                );
                fallback_summary(&inputs, ctx.token_budget)
            }
        }
    };

    let (node_entities, node_topics) = resolve_labels(strategy, &inputs, &output.content).await?;

    // Embed before any write so a failure aborts cleanly — same contract as
    // seal_one_level. No-provider configs seal embedding-less.
    //
    // #13021: skip when the embed input is empty/whitespace. `summarise()`
    // returns an empty content default when every input is whitespace, which
    // would otherwise round-trip a guaranteed 400 from the upstream embedding
    // API. The doc-subtree seal is sealed embedding-less, matching the
    // no-provider branch.
    let embed_input = truncate_for_embed(&output.content, 1_000);
    let embedding: Option<Vec<f32>> = if embed_input.trim().is_empty() {
        log::warn!(
            "[tree::bucket_seal] doc-subtree: empty summary content for tree_id={} level={} \
             — sealing without embedding (#13021)",
            tree.id,
            level
        );
        None
    } else {
        match build_write_embedder(config).context("build embedder during doc-subtree seal")? {
            None => None,
            Some(embedder) => {
                let v = embedder.embed(&embed_input).await.map_err(|e| {
                    let failure = crate::openhuman::memory_tree::health::classify_embed_error(&e);
                    anyhow::Error::new(failure).context(format!(
                        "embed doc-subtree summary tree_id={} level={}: {e:#}",
                        tree.id, level
                    ))
                })?;
                crate::openhuman::memory_tree::score::embed::pack_checked(&v).context(format!(
                    "doc-subtree embed dim check tree_id={} level={}",
                    tree.id, level
                ))?;
                Some(v)
            }
        }
    };

    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: tree.kind,
        level: target_level,
        parent_id: None,
        child_ids: child_ids.to_vec(),
        content: output.content,
        token_count: output.token_count,
        entities: node_entities,
        topics: node_topics,
        time_range_start,
        time_range_end,
        score,
        sealed_at: now,
        deleted: false,
        embedding,
        doc_id: doc_id.map(|s| s.to_string()),
        version_ms,
    };

    // Stage the .md file before opening the write tx (same fail-fast as
    // seal_one_level). Doc-subtree nodes land under
    // `source-<scope>/docs/<doc_slug>/v-<version_ms>/…` so the vault mirrors
    // the logical shape. Wikilink overrides are left unset.
    let tree_factory = TreeFactory::from_tree(tree);
    let summary_tree_kind = tree_factory.summary_tree_kind();
    let scope_slug = tree_factory.scope_slug();
    let compose_input = SummaryComposeInput {
        summary_id: &node.id,
        tree_kind: summary_tree_kind,
        tree_id: &node.tree_id,
        tree_scope: &tree.scope,
        level: node.level,
        child_ids: &node.child_ids,
        child_basenames: None,
        child_count: node.child_ids.len(),
        time_range_start: node.time_range_start,
        time_range_end: node.time_range_end,
        sealed_at: node.sealed_at,
        body: &node.content,
    };
    let content_root = config.memory_tree_content_root();
    if let Err(err) =
        crate::openhuman::memory_store::content::obsidian::ensure_obsidian_defaults(&content_root)
    {
        log::warn!(
            "[tree::bucket_seal] ensure_obsidian_defaults failed during doc-subtree seal: {err:#}"
        );
    }
    let doc_slug = doc_id.map(slugify_source_id);
    let layout = match doc_slug.as_deref() {
        Some(slug) => SummaryDiskLayout::DocSubtree {
            doc_slug: slug,
            version_ms,
        },
        None => SummaryDiskLayout::Standard,
    };
    let staged = stage_summary_with_layout(&content_root, &compose_input, &scope_slug, layout)
        .with_context(|| {
            format!(
                "stage_summary failed for doc-subtree node {}; seal aborted",
                node.id
            )
        })?;
    let summary_commit = summary_seal_batch(
        tree,
        "document_subtree_seal",
        std::slice::from_ref(&node),
        std::slice::from_ref(&staged.content_path),
    )
    .with_context(|| {
        format!(
            "build summary git batch failed for doc-subtree node {}; seal aborted",
            node.id
        )
    })?;

    // Persist the summary row + backlink children — NO buffer / tree-root
    // mutation (those belong to the merge tier).
    let node_for_tx = node.clone();
    let level_for_tx = level;
    let summary_id_for_tx = summary_id.clone();
    let signature = crate::openhuman::memory_store::chunks::store::tree_active_signature(config);
    with_connection(config, move |conn| {
        let tx = conn.unchecked_transaction()?;
        store::insert_summary_tx(&tx, &node_for_tx, Some(&staged), &signature)?;
        crate::openhuman::memory_tree::score::store::index_summary_entity_ids_tx(
            &tx,
            &node_for_tx.entities,
            &node_for_tx.id,
            node_for_tx.score,
            now.timestamp_millis(),
            Some(&node_for_tx.tree_id),
        )?;
        for child_id in &node_for_tx.child_ids {
            if level_for_tx == 0 {
                // Unconditional re-point (no `IS NULL` guard): a byte-identical
                // body chunk reused across doc versions upserts to the SAME row
                // (content-addressed id), so its single `parent_summary_id` must
                // follow the newest version. Doc subtrees seal newest-last, so
                // last-write-wins leaves the backlink on the latest doc-root —
                // the version retrieval surfaces — instead of stranding it on
                // the first (now-superseded) version's summary.
                tx.execute(
                    "UPDATE mem_tree_chunks SET parent_summary_id = ?1 \
                       WHERE id = ?2",
                    rusqlite::params![&summary_id_for_tx, child_id],
                )
                .context("backlink chunk to doc-subtree 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_tx, child_id],
                )
                .context("backlink summary to doc-subtree parent")?;
            }
        }
        tx.commit()?;
        Ok(())
    })?;

    commit_summary_seal(config, &summary_commit).with_context(|| {
        format!(
            "commit_summary_seal failed for doc-subtree node {} after DB seal commit",
            summary_id
        )
    })?;

    log::info!(
        "[tree::bucket_seal] doc-subtree sealed tree_id={} doc_id_hash={} level={}→{} summary_id={} children={}",
        tree.id,
        doc_id.map(redact).unwrap_or_default(),
        level,
        target_level,
        summary_id,
        child_ids.len()
    );

    Ok(node)
}

#[cfg(test)]
#[path = "bucket_seal_tests.rs"]
mod tests;
