Fix Gmail raw-backed memory tree sync (#3459)

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<String>,
+    thread: EmailThread,
+    raw_refs: Vec<RawRef>,
+) -> Result<IngestResult> {
+    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<i64>,
+    raw_refs_for_chunks: Option<Vec<RawRef>>,
 ) -> Result<IngestResult> {
     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<Option<usize>> {
         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

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<Option<Vec<RawRef>>> {

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<usize>) -> i64 {

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<DateTime<Utc>> {
-    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<DateTime<Utc>> {
     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::<i64>() {
             return DateTime::from_timestamp_millis(ms);
         }
@@ -254,10 +265,7 @@ pub fn parse_message_date(m: &Value) -> Option<DateTime<Utc>> {
             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());
     }
 

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<String>,
+}
+
 /// 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<usize> {
+    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<PageIngestOutcome> {
     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<String> = 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<EmailMessage> = 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<PageIngestOutcome> {
     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.

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<String> = 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 {

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<Mutex<()>> = 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 <ava@example.test>",
+            "to": ["Ben <ben@example.test>"],
+            "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 <ben@example.test>",
+            "to": ["Ava <ava@example.test>"],
+            "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();