From 8afc49d840e464f6385d7269367780f4b673da0c Mon Sep 17 00:00:00 2001 From: Steven Enamakel <31011319+senamakel@users.noreply.github.com> Date: Sat, 6 Jun 2026 19:49:28 -0400 Subject: [PATCH] Fix Gmail raw-backed memory tree sync (#3459) --- src/openhuman/memory/ingest_pipeline.rs | 37 ++++- src/openhuman/memory_store/chunks/raw_refs.rs | 14 +- src/openhuman/memory_store/chunks/store.rs | 3 +- .../memory_sync/canonicalize/email_clean.rs | 24 +++- .../composio/providers/gmail/ingest.rs | 104 ++++++++++---- .../composio/providers/gmail/provider.rs | 42 ++++-- .../memory_sync_providers_raw_coverage_e2e.rs | 133 +++++++++++++++++- 7 files changed, 309 insertions(+), 48 deletions(-) diff --git a/src/openhuman/memory/ingest_pipeline.rs b/src/openhuman/memory/ingest_pipeline.rs index b88868619..614bd63d1 100644 --- a/src/openhuman/memory/ingest_pipeline.rs +++ b/src/openhuman/memory/ingest_pipeline.rs @@ -13,7 +13,7 @@ use crate::core::event_bus::{publish_global, DomainEvent}; use crate::openhuman::config::Config; use crate::openhuman::memory::util::redact::redact; use crate::openhuman::memory_queue::{self as jobs, ExtractChunkPayload, NewJob}; -use crate::openhuman::memory_store::chunks::store as chunk_store; +use crate::openhuman::memory_store::chunks::store::{self as chunk_store, RawRef}; use crate::openhuman::memory_store::chunks::types::SourceKind; use crate::openhuman::memory_store::chunks::{chunk_markdown, ChunkerInput, ChunkerOptions}; use crate::openhuman::memory_store::content as content_store; @@ -88,7 +88,7 @@ pub async fn ingest_chat( Some(c) => c, None => return Ok(IngestResult::empty(source_id)), }; - persist(config, source_id, canonical, None).await + persist(config, source_id, canonical, None, None).await } /// Ingest an email thread: canonicalise → chunk → fast-score → persist → enqueue @@ -110,7 +110,28 @@ pub async fn ingest_email( Some(c) => c, None => return Ok(IngestResult::empty(source_id)), }; - persist(config, source_id, canonical, None).await + persist(config, source_id, canonical, None, None).await +} + +/// Ingest an email thread whose chunk bodies are backed by a raw archive file. +/// +/// Raw refs are committed in the same transaction as the chunk rows and +/// `extract_chunk` jobs, so workers can never observe a raw-backed email chunk +/// without a resolvable body source. +pub async fn ingest_email_with_raw_refs( + config: &Config, + source_id: &str, + owner: &str, + tags: Vec, + thread: EmailThread, + raw_refs: Vec, +) -> Result { + let canonical = + match email::canonicalise(source_id, owner, &tags, thread).map_err(anyhow::Error::msg)? { + Some(c) => c, + None => return Ok(IngestResult::empty(source_id)), + }; + persist(config, source_id, canonical, None, Some(raw_refs)).await } /// Ingest a single document: canonicalise → chunk → fast-score → persist → @@ -175,7 +196,7 @@ pub async fn ingest_document_versioned( Some(c) => c, None => return Ok(IngestResult::empty(source_id)), }; - persist(config, source_id, canonical, version_ms).await + persist(config, source_id, canonical, version_ms, None).await } /// Best-effort pre-canonicalisation check. The transactional claim inside @@ -198,6 +219,7 @@ async fn persist( source_id: &str, canonical: CanonicalisedSource, gate_version_ms: Option, + raw_refs_for_chunks: Option>, ) -> Result { let source_kind_for_store = canonical.metadata.source_kind; @@ -266,6 +288,7 @@ async fn persist( let staged_for_store = staged.clone(); let results_for_store = all_results.clone(); let source_id_for_store = source_id.to_string(); + let raw_refs_for_store = raw_refs_for_chunks.clone(); let written = tokio::task::spawn_blocking(move || -> Result> { use std::collections::{HashMap, HashSet}; chunk_store::with_connection(&config_owned, |conn| { @@ -347,6 +370,12 @@ async fn persist( let n = chunk_store::upsert_staged_chunks_tx(&tx, &staged_for_store)?; + if let Some(ref refs) = raw_refs_for_store { + for s in &staged_for_store { + chunk_store::set_chunk_raw_refs_tx(&tx, &s.chunk.id, refs)?; + } + } + // Re-ingest of identical content (same chunk_id) must NOT // downgrade chunks that have already progressed through the // async pipeline. Without this guard, a re-ingest would reset diff --git a/src/openhuman/memory_store/chunks/raw_refs.rs b/src/openhuman/memory_store/chunks/raw_refs.rs index 5d581d6e1..1eccdcb8a 100644 --- a/src/openhuman/memory_store/chunks/raw_refs.rs +++ b/src/openhuman/memory_store/chunks/raw_refs.rs @@ -6,7 +6,7 @@ //! directly instead of going through the SQL preview path. use anyhow::{Context, Result}; -use rusqlite::{params, OptionalExtension}; +use rusqlite::{params, OptionalExtension, Transaction}; use super::with_connection; use crate::openhuman::config::Config; @@ -45,6 +45,18 @@ pub fn set_chunk_raw_refs(config: &Config, chunk_id: &str, refs: &[RawRef]) -> R }) } +/// Stash raw archive pointers on a chunk row inside a caller-owned +/// transaction. Used by ingest producers so raw-backed chunks are readable +/// before their `extract_chunk` jobs are committed. +pub fn set_chunk_raw_refs_tx(tx: &Transaction<'_>, chunk_id: &str, refs: &[RawRef]) -> Result<()> { + let json = serde_json::to_string(refs).context("serialize raw_refs")?; + tx.execute( + "UPDATE mem_tree_chunks SET raw_refs_json = ?1 WHERE id = ?2", + params![json, chunk_id], + )?; + Ok(()) +} + /// Return the raw-archive pointers stored in SQLite for `chunk_id`, /// or `None` if no `raw_refs_json` was recorded. pub fn get_chunk_raw_refs(config: &Config, chunk_id: &str) -> Result>> { diff --git a/src/openhuman/memory_store/chunks/store.rs b/src/openhuman/memory_store/chunks/store.rs index 2fc3637fc..5f19fdf32 100644 --- a/src/openhuman/memory_store/chunks/store.rs +++ b/src/openhuman/memory_store/chunks/store.rs @@ -1186,7 +1186,8 @@ use migrations::{migrate_legacy_embeddings_to_sidecar, purge_global_topic_trees} mod raw_refs; pub use raw_refs::{ get_chunk_content_path, get_chunk_content_pointers, get_chunk_raw_refs, - get_summary_content_pointers, list_summaries_with_content_path, set_chunk_raw_refs, RawRef, + get_summary_content_pointers, list_summaries_with_content_path, set_chunk_raw_refs, + set_chunk_raw_refs_tx, RawRef, }; fn normalized_limit(requested: Option) -> i64 { diff --git a/src/openhuman/memory_sync/canonicalize/email_clean.rs b/src/openhuman/memory_sync/canonicalize/email_clean.rs index 766c38291..8c261eabe 100644 --- a/src/openhuman/memory_sync/canonicalize/email_clean.rs +++ b/src/openhuman/memory_sync/canonicalize/email_clean.rs @@ -223,13 +223,24 @@ fn strip_day_of_week_prefix(s: &str) -> Option<&str> { /// on the raw `serde_json::Value` so callers that work off the slim /// envelope JSON don't have to reshape it first. pub fn parse_message_date(m: &Value) -> Option> { - let raw = m.get("date")?; + if let Some(dt) = m.get("date").and_then(parse_date_value) { + return Some(dt); + } + if let Some(dt) = m.get("internalDate").and_then(parse_date_value) { + return Some(dt); + } + m.get("data") + .and_then(|data| data.get("internalDate")) + .and_then(parse_date_value) +} + +fn parse_date_value(raw: &Value) -> Option> { if let Some(s) = raw.as_str() { let s = s.trim(); if s.is_empty() { return None; } - // Epoch millis as a string? + // Epoch millis as a string? Gmail's `internalDate` uses this form. if let Ok(ms) = s.parse::() { return DateTime::from_timestamp_millis(ms); } @@ -254,10 +265,7 @@ pub fn parse_message_date(m: &Value) -> Option> { return d.and_hms_opt(0, 0, 0).map(|n| n.and_utc()); } } - if let Some(ms) = raw.as_i64() { - return DateTime::from_timestamp_millis(ms); - } - None + raw.as_i64().and_then(DateTime::from_timestamp_millis) } #[cfg(test)] @@ -370,11 +378,15 @@ mod tests { let rfc = json!({"date": "Mon, 21 Apr 2026 10:00:00 +0000"}); let ms = json!({"date": 1745236800000_i64}); let ms_str = json!({"date": "1745236800000"}); + let internal_ms_str = json!({"internalDate": "1745236800000"}); + let nested_internal_ms_str = json!({"data": {"internalDate": "1745236800000"}}); let date_only = json!({"date": "2026-04-21"}); assert!(parse_message_date(&iso).is_some()); assert!(parse_message_date(&rfc).is_some()); assert!(parse_message_date(&ms).is_some()); assert!(parse_message_date(&ms_str).is_some()); + assert!(parse_message_date(&internal_ms_str).is_some()); + assert!(parse_message_date(&nested_internal_ms_str).is_some()); assert!(parse_message_date(&date_only).is_some()); } diff --git a/src/openhuman/memory_sync/composio/providers/gmail/ingest.rs b/src/openhuman/memory_sync/composio/providers/gmail/ingest.rs index b7949c4fc..161e3335d 100644 --- a/src/openhuman/memory_sync/composio/providers/gmail/ingest.rs +++ b/src/openhuman/memory_sync/composio/providers/gmail/ingest.rs @@ -21,9 +21,11 @@ use anyhow::Result; use serde_json::Value; use crate::openhuman::config::Config; -use crate::openhuman::memory::ingest_pipeline::{ingest_email, IngestResult}; +use crate::openhuman::memory::ingest_pipeline::{ + ingest_email, ingest_email_with_raw_refs, IngestResult, +}; use crate::openhuman::memory::util::redact::redact; -use crate::openhuman::memory_store::chunks::store::{set_chunk_raw_refs, RawRef}; +use crate::openhuman::memory_store::chunks::store::RawRef; use crate::openhuman::memory_store::content::raw::{ self as raw_store, raw_rel_path, slug_account_email, RawItem, RawKind, }; @@ -38,6 +40,12 @@ pub const GMAIL_PROVIDER: &str = "gmail"; /// callers filter on these. pub const DEFAULT_TAGS: &[&str] = &["gmail", "ingested"]; +#[derive(Debug, Clone, Default)] +pub struct PageIngestOutcome { + pub chunks_written: usize, + pub item_ids_ingested: Vec, +} + /// Group raw page messages by the sorted set of distinct participants /// (`from` ∪ `to`-list). CC is deliberately excluded from the bucket key /// so CC-only recipients don't fragment conversations. All messages @@ -238,8 +246,21 @@ pub async fn ingest_page_into_memory_tree( account_email: Option<&str>, page_messages: &[Value], ) -> Result { + Ok( + ingest_page_into_memory_tree_with_outcome(config, owner, account_email, page_messages) + .await? + .chunks_written, + ) +} + +pub async fn ingest_page_into_memory_tree_with_outcome( + config: &Config, + owner: &str, + account_email: Option<&str>, + page_messages: &[Value], +) -> Result { if page_messages.is_empty() { - return Ok(0); + return Ok(PageIngestOutcome::default()); } let account_source_id = account_email .filter(|e| !e.trim().is_empty()) @@ -248,13 +269,17 @@ pub async fn ingest_page_into_memory_tree( // Best-effort raw archive — runs once per page, before chunking, so // a chunker bug doesn't block us from capturing the source bytes. if let Some(ref source_id) = account_source_id { - if let Err(e) = write_raw_archive(config, source_id, page_messages) { + write_raw_archive(config, source_id, page_messages).map_err(|e| { log::warn!( "[composio:gmail][ingest] raw archive write failed source_id_hash={} err={:#}", redact(source_id), e ); - } + anyhow::anyhow!( + "gmail raw archive write failed for {}: {e:#}", + redact(source_id) + ) + })?; } // Per-account ingest path: one ingest call per upstream message so @@ -265,12 +290,14 @@ pub async fn ingest_page_into_memory_tree( // legacy participant-bucket path when we can't derive an // account-scoped source id (CLI runs / missing profile fetch). if let Some(ref source_id) = account_source_id { - let total_chunks = ingest_per_message(config, source_id, owner, page_messages).await; + let outcome = ingest_per_message(config, source_id, owner, page_messages).await?; log::info!( - "[composio:gmail][ingest] page_done owner_hash={} chunks={total_chunks} mode=per-account", + "[composio:gmail][ingest] page_done owner_hash={} chunks={} items={} mode=per-account", redact(owner), + outcome.chunks_written, + outcome.item_ids_ingested.len(), ); - return Ok(total_chunks); + return Ok(outcome); } // Legacy fallback: participant-bucketed thread ingest. No @@ -280,7 +307,17 @@ pub async fn ingest_page_into_memory_tree( let buckets = bucket_by_participants(page_messages); let mut total_chunks = 0usize; let mut total_buckets = 0usize; + let mut item_ids_ingested = Vec::new(); for (participants, raw_msgs) in &buckets { + let ids_in_bucket: Vec = raw_msgs + .iter() + .filter_map(|raw| { + raw.get("id") + .and_then(|v| v.as_str()) + .filter(|s| !s.is_empty()) + .map(str::to_string) + }) + .collect(); let messages: Vec = raw_msgs .iter() .filter_map(|raw| raw_to_email_message(raw)) @@ -304,6 +341,7 @@ pub async fn ingest_page_into_memory_tree( Ok(IngestResult { chunks_written, .. }) => { total_chunks += chunks_written; total_buckets += 1; + item_ids_ingested.extend(ids_in_bucket); } Err(e) => { log::warn!( @@ -319,7 +357,10 @@ pub async fn ingest_page_into_memory_tree( "[composio:gmail][ingest] page_done owner_hash={} buckets={total_buckets} chunks={total_chunks} mode=per-participants", redact(owner), ); - Ok(total_chunks) + Ok(PageIngestOutcome { + chunks_written: total_chunks, + item_ids_ingested, + }) } /// Per-account ingest: one `ingest_email` call per upstream message. @@ -335,8 +376,9 @@ async fn ingest_per_message( source_id: &str, owner: &str, page_messages: &[Value], -) -> usize { +) -> Result { let mut total_chunks = 0usize; + let mut item_ids_ingested = Vec::new(); for raw in page_messages { let id = raw .get("id") @@ -346,6 +388,17 @@ async fn ingest_per_message( let Some(sent_at) = parse_message_date(raw) else { continue; }; + let has_raw_body = raw + .get("markdown") + .and_then(|v| v.as_str()) + .is_some_and(|body| !body.trim().is_empty()); + if !has_raw_body { + log::debug!( + "[composio:gmail][ingest] skipping empty raw-backed message msg_id_hash={}", + redact(msg_id) + ); + continue; + } let Some(message) = raw_to_email_message(raw) else { continue; }; @@ -364,22 +417,16 @@ async fn ingest_per_message( messages: vec![message], }; let tags = DEFAULT_TAGS.iter().map(|s| (*s).to_string()).collect(); - match ingest_email(config, source_id, owner, tags, thread).await { + let refs = vec![RawRef { + path: raw_path.clone(), + start: 0, + end: None, + }]; + match ingest_email_with_raw_refs(config, source_id, owner, tags, thread, refs).await { Ok(result) => { total_chunks += result.chunks_written; - let refs = vec![RawRef { - path: raw_path.clone(), - start: 0, - end: None, - }]; - for chunk_id in &result.chunk_ids { - if let Err(e) = set_chunk_raw_refs(config, chunk_id, &refs) { - log::warn!( - "[composio:gmail][ingest] set_chunk_raw_refs failed chunk_id={} err={:#}", - chunk_id, - e - ); - } + if result.chunks_written > 0 || !result.chunk_ids.is_empty() { + item_ids_ingested.push(msg_id.to_string()); } } Err(e) => { @@ -388,10 +435,17 @@ async fn ingest_per_message( redact(msg_id), e ); + return Err(anyhow::anyhow!( + "gmail per-message ingest failed msg_id_hash={}: {e:#}", + redact(msg_id) + )); } } } - total_chunks + Ok(PageIngestOutcome { + chunks_written: total_chunks, + item_ids_ingested, + }) } /// Mirror a page of raw Gmail messages into the on-disk raw archive. diff --git a/src/openhuman/memory_sync/composio/providers/gmail/provider.rs b/src/openhuman/memory_sync/composio/providers/gmail/provider.rs index bcc57199a..ec1051830 100644 --- a/src/openhuman/memory_sync/composio/providers/gmail/provider.rs +++ b/src/openhuman/memory_sync/composio/providers/gmail/provider.rs @@ -25,7 +25,7 @@ use async_trait::async_trait; use serde_json::{json, Value}; -use super::ingest::ingest_page_into_memory_tree; +use super::ingest::ingest_page_into_memory_tree_with_outcome; use super::sync; use crate::openhuman::memory_sync::composio::providers::sync_state::{extract_item_id, SyncState}; use crate::openhuman::memory_sync::composio::providers::{ @@ -312,6 +312,7 @@ impl ComposioProvider for GmailProvider { let mut page_token: Option = None; let mut stop_reason: &'static str = "max_pages"; let mut hit_cap_boundary = false; + let mut had_ingest_failures = false; for page_num in 0..max_pages { if state.budget_exhausted() { @@ -505,7 +506,7 @@ impl ComposioProvider for GmailProvider { // the storage layer. if !new_messages.is_empty() { let owner = format!("gmail-sync:{connection_id}"); - match ingest_page_into_memory_tree( + match ingest_page_into_memory_tree_with_outcome( ctx.config.as_ref(), &owner, account_email.as_deref(), @@ -513,24 +514,43 @@ impl ComposioProvider for GmailProvider { ) .await { - Ok(n) => { - for id in &pending_synced_ids { + Ok(outcome) => { + let ingested_ids: std::collections::BTreeSet<&str> = outcome + .item_ids_ingested + .iter() + .map(String::as_str) + .collect(); + for id in pending_synced_ids + .iter() + .filter(|id| ingested_ids.contains(id.as_str())) + { state.mark_synced(id); } // total_persisted tracks messages, not chunks, for // metric stability with the previous per-message // persist path. n is the chunk count which we log // for diagnostic purposes only. - total_persisted += new_messages.len(); - cap.record(new_messages.len()); + total_persisted += outcome.item_ids_ingested.len(); + cap.record(outcome.item_ids_ingested.len()); tracing::debug!( page = page_num, new_messages = new_messages.len(), - ingested_chunks = n, + ingested_messages = outcome.item_ids_ingested.len(), + ingested_chunks = outcome.chunks_written, "[composio:gmail] page ingested into memory tree" ); + if outcome.item_ids_ingested.len() < new_messages.len() { + tracing::warn!( + page = page_num, + new_messages = new_messages.len(), + ingested_messages = outcome.item_ids_ingested.len(), + "[composio:gmail] partial page ingest; holding cursor for retry" + ); + had_ingest_failures = true; + } } Err(e) => { + had_ingest_failures = true; tracing::warn!( error = %format!("{e:#}"), page = page_num, @@ -575,15 +595,17 @@ impl ComposioProvider for GmailProvider { // ── Step 5: advance cursor and save state ─────────────────── // Hold the cursor on a cap-truncated pass so the next sync re-scans the unseen tail. - if !hit_cap_boundary { + if !had_ingest_failures && !hit_cap_boundary { if let Some(new_cursor) = newest_date { state.advance_cursor(&new_cursor); } } else { tracing::warn!( connection_id = %connection_id, - "[composio:gmail] holding cursor — cap-truncated pass; next sync will re-scan \ - the unseen tail" + had_ingest_failures, + hit_cap_boundary, + "[composio:gmail] holding cursor — ingest failures or cap-truncated pass; next sync \ + will re-scan the affected range" ); } if let Some(ref freshest) = newest_id { diff --git a/tests/memory_sync_providers_raw_coverage_e2e.rs b/tests/memory_sync_providers_raw_coverage_e2e.rs index fa7d0c57b..9cc9c9fd6 100644 --- a/tests/memory_sync_providers_raw_coverage_e2e.rs +++ b/tests/memory_sync_providers_raw_coverage_e2e.rs @@ -19,6 +19,15 @@ use openhuman_core::openhuman::credentials::{ AuthService, APP_SESSION_PROVIDER, DEFAULT_AUTH_PROFILE_NAME, }; use openhuman_core::openhuman::memory::global as memory_global; +use openhuman_core::openhuman::memory::jobs::drain_until_idle; +use openhuman_core::openhuman::memory::tree_source::get_or_create_source_tree; +use openhuman_core::openhuman::memory_store::chunks::store::{ + count_chunks_by_lifecycle_status, get_chunk_raw_refs, list_chunks, ListChunksQuery, + CHUNK_STATUS_BUFFERED, +}; +use openhuman_core::openhuman::memory_store::chunks::types::SourceKind; +use openhuman_core::openhuman::memory_store::content::read::read_chunk_body; +use openhuman_core::openhuman::memory_store::trees::store as tree_store; use openhuman_core::openhuman::memory_sync::composio::bus::{ ComposioConfigChangedSubscriber, ComposioConnectionCreatedSubscriber, ComposioTriggerSubscriber, }; @@ -35,6 +44,7 @@ use openhuman_core::openhuman::memory_sync::composio::providers::slack::{ use openhuman_core::openhuman::memory_sync::composio::providers::{ ComposioProvider, ProviderContext, SyncReason, TaskFetchFilter, }; +use openhuman_core::openhuman::memory_tree::tree::bucket_seal::LabelStrategy; static ENV_LOCK: OnceLock> = OnceLock::new(); @@ -367,7 +377,11 @@ async fn gmail_ingest_archives_account_messages_and_legacy_participant_buckets() let raw_root = config.memory_tree_content_root().join("raw"); let archived: Vec<_> = walk_files(&raw_root) .into_iter() - .filter(|p| p.to_string_lossy().contains("gmail-round17-a")) + .filter(|p| { + p.file_name() + .and_then(|name| name.to_str()) + .is_some_and(|name| name.contains("gmail-round17-a")) + }) .collect(); assert_eq!(archived.len(), 1, "raw archive should include message a"); let archived_body = std::fs::read_to_string(&archived[0]).expect("archived body"); @@ -401,6 +415,123 @@ async fn gmail_ingest_archives_account_messages_and_legacy_participant_buckets() assert!(legacy >= 1, "orphan fallback bucket should ingest"); } +#[tokio::test] +async fn gmail_raw_backed_messages_drain_into_source_tree_summary() { + let _guard = env_lock(); + let tmp = TempDir::new().expect("tempdir"); + let _workspace = EnvGuard::set_path("OPENHUMAN_WORKSPACE", tmp.path()); + let _home = EnvGuard::set_path("HOME", tmp.path()); + let _backend = EnvGuard::unset("BACKEND_URL"); + let config = config_in(&tmp); + persist_config(&config).await; + + let page = vec![ + json!({ + "id": "gmail-tree-a", + "from": "Ava ", + "to": ["Ben "], + "subject": "Phoenix launch plan", + "date": "2026-05-29T10:00:00Z", + "markdown": "Phoenix migration launches Friday. Ava owns rollout validation and Ben owns customer notices." + }), + json!({ + "id": "gmail-tree-b", + "from": "Ben ", + "to": ["Ava "], + "subject": "Re: Phoenix launch plan", + "date": "2026-05-29T10:05:00Z", + "markdown": "Confirmed. Customer notices go out after staging checks and the rollback doc is reviewed." + }), + ]; + + let outcome = gmail_ingest::ingest_page_into_memory_tree_with_outcome( + &config, + "owner-gmail-tree", + Some("flow@example.test"), + &page, + ) + .await + .expect("gmail ingest outcome"); + + assert_eq!( + outcome.item_ids_ingested, + vec!["gmail-tree-a".to_string(), "gmail-tree-b".to_string()] + ); + assert!( + outcome.chunks_written >= 2, + "expected one or more chunks per message" + ); + + let source_id = "gmail:flow-at-example-dot-test"; + let chunks = list_chunks( + &config, + &ListChunksQuery { + source_kind: Some(SourceKind::Email), + source_id: Some(source_id.to_string()), + limit: Some(10), + ..Default::default() + }, + ) + .expect("list gmail chunks"); + assert_eq!(chunks.len(), outcome.chunks_written); + + for chunk in &chunks { + let refs = get_chunk_raw_refs(&config, &chunk.id) + .expect("raw refs lookup") + .expect("raw refs must be set before extract can run"); + assert_eq!(refs.len(), 1); + assert!( + refs[0].path.contains("gmail-tree-"), + "raw ref should point at the source message file: {:?}", + refs[0].path + ); + let full_body = read_chunk_body(&config, &chunk.id).expect("read raw-backed chunk body"); + assert!( + full_body.contains("Phoenix") || full_body.contains("Customer notices"), + "chunk body should hydrate from raw archive, got: {full_body}" + ); + } + + drain_until_idle(&config) + .await + .expect("extract and append jobs should drain"); + let buffered = + count_chunks_by_lifecycle_status(&config, CHUNK_STATUS_BUFFERED).expect("buffered count"); + assert_eq!(buffered, outcome.chunks_written as u64); + + let tree = get_or_create_source_tree(&config, source_id).expect("source tree"); + let l0 = tree_store::get_buffer(&config, &tree.id, 0).expect("source L0 buffer"); + assert_eq!( + l0.item_ids.len(), + outcome.chunks_written, + "all Gmail chunks should reach the source tree buffer" + ); + + let sealed = openhuman_core::openhuman::memory_tree::tree::flush::flush_stale_buffers( + &config, + chrono::Duration::zero(), + &LabelStrategy::Empty, + ) + .await + .expect("force flush gmail source tree"); + assert!(sealed > 0, "low-volume Gmail source should seal on flush"); + + let l1 = + tree_store::list_summaries_at_level(&config, &tree.id, 1).expect("list source summaries"); + assert!( + !l1.is_empty(), + "Gmail source tree should have a sealed summary after flush" + ); + let summary_body = openhuman_core::openhuman::memory_store::content::read::read_summary_body( + &config, &l1[0].id, + ) + .expect("read gmail summary body"); + assert!( + summary_body.contains("Phoenix") || summary_body.contains("Customer"), + "summary should preserve Gmail content, got: {summary_body}" + ); +} + #[tokio::test] async fn slack_provider_profile_postprocess_trigger_and_ingest_use_loopback_composio() { let _guard = env_lock();