diff --git a/src/openhuman/memory_sync/composio/providers/gmail/mod.rs b/src/openhuman/memory_sync/composio/providers/gmail/mod.rs index a20f787d9..edc27ef37 100644 --- a/src/openhuman/memory_sync/composio/providers/gmail/mod.rs +++ b/src/openhuman/memory_sync/composio/providers/gmail/mod.rs @@ -1,6 +1,7 @@ pub mod ingest; mod post_process; mod provider; +mod source; mod sync; #[cfg(test)] mod tests; diff --git a/src/openhuman/memory_sync/composio/providers/gmail/provider.rs b/src/openhuman/memory_sync/composio/providers/gmail/provider.rs index ec1051830..bcb5d50a2 100644 --- a/src/openhuman/memory_sync/composio/providers/gmail/provider.rs +++ b/src/openhuman/memory_sync/composio/providers/gmail/provider.rs @@ -25,16 +25,14 @@ use async_trait::async_trait; use serde_json::{json, Value}; -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 super::source::run_gmail_sync; use crate::openhuman::memory_sync::composio::providers::{ pick_str, resolve_sync_interval_secs, ComposioProvider, CuratedTool, ProviderContext, ProviderUserProfile, SyncOutcome, SyncReason, }; -const ACTION_GET_PROFILE: &str = "GMAIL_GET_PROFILE"; -const ACTION_FETCH_EMAILS: &str = "GMAIL_FETCH_EMAILS"; +pub(super) const ACTION_GET_PROFILE: &str = "GMAIL_GET_PROFILE"; +pub(super) const ACTION_FETCH_EMAILS: &str = "GMAIL_FETCH_EMAILS"; /// Base Gmail search query used on every sync pass. /// @@ -50,48 +48,6 @@ pub(super) const BASE_QUERY: &str = "-in:spam -in:trash"; /// to fetch outbound messages. Exported `pub(super)` for use in regression tests. pub(super) const SENT_QUERIES: &[&str] = &["from:me", "label:SENT", "in:sent"]; -/// Page size per API call. Kept moderate so each call is fast and we -/// get frequent checkpoints for the daily budget. -const PAGE_SIZE: u32 = 25; - -/// Larger page size for the very first sync after OAuth so the user -/// gets a meaningful initial snapshot. -const INITIAL_PAGE_SIZE: u32 = 50; - -/// Maximum pages to fetch in a single sync pass (guards against infinite -/// pagination loops). Combined with PAGE_SIZE this yields at most -/// 500 items per sync pass, well within the daily budget. -const MAX_PAGES_PER_SYNC: u32 = 20; - -/// Adaptive page cap applied when a successful sync ran very recently. -/// If the previous sync wrote within -/// [`RECENT_SYNC_WINDOW_MS`], the upcoming sync is unlikely to need more -/// than a couple of pages — anything beyond that is almost certainly -/// re-fetching content `synced_ids` will throw away anyway. -const RECENT_SYNC_MAX_PAGES: u32 = 2; - -/// "Recent" window used by the adaptive page cap. Five minutes is short -/// enough that periodic-tick churn and trigger-driven retries fall -/// inside it, but long enough that a genuine "no-activity" gap (e.g. -/// the user closing the laptop) drops back to the full -/// `MAX_PAGES_PER_SYNC` ceiling on the next wake. -const RECENT_SYNC_WINDOW_MS: u64 = 5 * 60 * 1000; - -/// Paths to try when extracting a message's unique ID from the Composio -/// response envelope. -const MESSAGE_ID_PATHS: &[&str] = &["id", "data.id", "messageId", "data.messageId"]; - -/// Paths for extracting the internal date (epoch millis or date string) -/// used as the sync cursor. -const MESSAGE_DATE_PATHS: &[&str] = &[ - "internalDate", - "data.internalDate", - "date", - "data.date", - "receivedAt", - "data.receivedAt", -]; - pub struct GmailProvider; impl GmailProvider { @@ -192,486 +148,14 @@ impl ComposioProvider for GmailProvider { Ok(profile) } + /// Incremental sync via the generic + /// [`orchestrator`](crate::openhuman::memory_sync::composio::providers::orchestrator): + /// pagination, dedup, the `max_items` cap, and cursor handling live in + /// `run_sync`; the Gmail-specific primitives — the account-email preamble, + /// server-side `after:` depth window, adaptive page ceiling, all-synced + /// stop, and batch ingest — live in [`super::source`]. async fn sync(&self, ctx: &ProviderContext, reason: SyncReason) -> Result { - let started_at_ms = sync::now_ms(); - let connection_id = ctx - .connection_id - .clone() - .unwrap_or_else(|| "default".to_string()); - - tracing::info!( - connection_id = %connection_id, - reason = reason.as_str(), - "[composio:gmail] incremental sync starting" - ); - - // ── Step 1: load persistent sync state ────────────────────── - let Some(memory) = ctx.memory_client() else { - return Err("[composio:gmail] memory client not ready".to_string()); - }; - let mut state = SyncState::load(&memory, "gmail", &connection_id).await?; - - // Fetch the account email up-front so every chunk gets a stable - // per-account `source_id` (`gmail:{slug(email)}`). One HTTP - // round-trip per sync; if it fails we fall back to the legacy - // per-participants bucketing inside the ingest call so we - // still write *something* useful. - let account_email: Option = match self.fetch_user_profile(ctx).await { - Ok(profile) => profile.email, - Err(e) => { - tracing::warn!( - connection_id = %connection_id, - error = ?e, - "[composio:gmail] fetch_user_profile failed; ingest will fall back to per-participants source_id" - ); - None - } - }; - - // ── Step 2: check daily budget ────────────────────────────── - if state.budget_exhausted() { - tracing::info!( - connection_id = %connection_id, - "[composio:gmail] daily request budget exhausted, skipping sync" - ); - return Ok(SyncOutcome { - toolkit: "gmail".to_string(), - connection_id: Some(connection_id), - reason: reason.as_str().to_string(), - items_ingested: 0, - started_at_ms, - finished_at_ms: sync::now_ms(), - summary: "gmail sync skipped: daily budget exhausted".to_string(), - details: json!({ "budget_exhausted": true }), - }); - } - - // ── Step 3: paginated incremental fetch ───────────────────── - let page_size = match reason { - SyncReason::ConnectionCreated => INITIAL_PAGE_SIZE, - _ => PAGE_SIZE, - }; - - // Adaptive page cap: if the previous successful sync wrote - // within the recent window, cap pagination aggressively. - // Initial backfills (`ConnectionCreated`) skip the cap — they - // legitimately want the larger ceiling — and the cap only - // kicks in when we have a prior `last_sync_at_ms` to compare - // against, so first-ever syncs are unaffected. - let base_max_pages = match reason { - SyncReason::ConnectionCreated => MAX_PAGES_PER_SYNC, - _ => match state.last_sync_at_ms { - Some(last_ms) if sync::now_ms().saturating_sub(last_ms) < RECENT_SYNC_WINDOW_MS => { - tracing::debug!( - connection_id = %connection_id, - last_sync_at_ms = last_ms, - cap = RECENT_SYNC_MAX_PAGES, - "[composio:gmail] recent sync — applying adaptive page cap" - ); - RECENT_SYNC_MAX_PAGES - } - _ => MAX_PAGES_PER_SYNC, - }, - }; - - // ctx.max_items: route through ItemCap so the page ceiling, mid-page - // clamp, and post-page hard stop all share one source of truth. - let mut cap = super::super::helpers::ItemCap::new(ctx.max_items); - let max_pages = cap.max_pages(page_size, base_max_pages); - if ctx.max_items.is_some() && max_pages < base_max_pages { - tracing::debug!( - connection_id = %connection_id, - max_items = ?ctx.max_items, - page_size, - effective_max_pages = max_pages, - "[composio:gmail] [memory_sync] applying max_items page cap from source config" - ); - } - - // ctx.sync_depth_days: on first sync (no cursor), add an after: floor. - let depth_floor_filter: Option = if state.cursor.is_none() { - ctx.sync_depth_days.map(|days| { - let floor_secs = super::super::helpers::epoch_floor_from_depth(days); - tracing::debug!( - connection_id = %connection_id, - sync_depth_days = days, - floor_epoch_secs = floor_secs, - "[composio:gmail] [memory_sync] applying sync_depth_days floor on first sync" - ); - floor_secs.to_string() - }) - } else { - None - }; - - let mut total_fetched: usize = 0; - let mut total_persisted: usize = 0; - let mut total_requests: u32 = 0; - let mut newest_date: Option = None; - let mut newest_id: Option = None; - 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() { - tracing::info!( - page = page_num, - "[composio:gmail] budget exhausted mid-sync, stopping pagination" - ); - stop_reason = "budget_exhausted"; - break; - } - - // Build the Gmail query. Prefer second-precision - // `after:` over the old day-level `after:YYYY/MM/DD` - // so same-day re-ticks do not re-fetch a whole day's - // window every time. Fall back to the day filter only when - // the cursor cannot be parsed as a timestamp. - // - // NOTE: We intentionally do NOT restrict to `in:inbox` here. - // The original query `in:inbox -in:spam -in:trash` meant sent - // emails (label:SENT) were never fetched and therefore the - // agent could not answer questions about outbound mail (issue #1713). - // Removing `in:inbox` lets Gmail return both inbox and sent - // messages while still excluding spam and trash. - let mut query = BASE_QUERY.to_string(); - if let Some(ref cursor) = state.cursor { - if let Some(epoch_filter) = sync::cursor_to_gmail_after_epoch_filter(cursor) { - query.push_str(&format!(" after:{epoch_filter}")); - tracing::debug!( - page = page_num, - filter = %epoch_filter, - "[composio:gmail] using epoch filter from cursor" - ); - } else if let Some(date_filter) = sync::cursor_to_gmail_after_filter(cursor) { - query.push_str(&format!(" after:{date_filter}")); - tracing::debug!( - page = page_num, - filter = %date_filter, - "[composio:gmail] using day-level filter from cursor (epoch parse failed)" - ); - } - } else if let Some(ref floor) = depth_floor_filter { - // First sync with sync_depth_days: apply the epoch floor. - query.push_str(&format!(" after:{floor}")); - } - - let mut args = json!({ - "max_results": page_size, - "query": query, - }); - if let Some(ref token) = page_token { - args["page_token"] = json!(token); - } - - let mut resp = ctx - .execute(ACTION_FETCH_EMAILS, Some(args.clone())) - .await - .map_err(|e| { - format!("[composio:gmail] {ACTION_FETCH_EMAILS} page {page_num}: {e:#}") - })?; - - state.record_requests(1); - total_requests += 1; - - if !resp.successful { - let err = resp - .error - .clone() - .unwrap_or_else(|| "provider reported failure".to_string()); - // Save state so budget accounting isn't lost. - let _ = state.save(&memory).await; - return Err(format!( - "[composio:gmail] {ACTION_FETCH_EMAILS} page {page_num}: {err}" - )); - } - - // ── Step 4: pull the backend's pre-rendered `markdownFormatted` - // onto each message so the raw archive sees URL-shortened, - // footer-stripped output. Done BEFORE post_process so the - // reshape can pick up the per-message field. Then run the - // usual post-process which slims the envelope and feeds - // `extract_markdown_body` (which now prefers - // `markdownFormatted` per message). - if let Some(top_md) = resp - .markdown_formatted - .as_deref() - .map(str::trim) - .filter(|s| !s.is_empty()) - { - super::post_process::apply_response_level_markdown(&mut resp.data, top_md); - } - self.post_process_action_result(ACTION_FETCH_EMAILS, Some(&args), &mut resp.data); - - let messages = sync::extract_messages(&resp.data); - total_fetched += messages.len(); - - if messages.is_empty() { - tracing::debug!( - page = page_num, - "[composio:gmail] empty page, stopping pagination" - ); - stop_reason = "empty_page"; - break; - } - - // First-message early-stop: when the very first message of - // the very first page matches the id we recorded at the - // end of the previous sync, the inbox has not changed and - // there is nothing left to fetch. Saves up to N-1 wasted - // pages on quiet inboxes where the day-level filter would - // otherwise re-fetch the same window. - if page_num == 0 { - let first_id = messages - .first() - .and_then(|m| extract_item_id(m, MESSAGE_ID_PATHS)); - if let (Some(seen), Some(first)) = - (state.last_seen_id.as_deref(), first_id.as_deref()) - { - if seen == first { - tracing::debug!( - connection_id = %connection_id, - first_id = %first, - "[composio:gmail] first page head matches last_seen_id — no new mail" - ); - stop_reason = "head_unchanged"; - // Capture the same id as the newest so the - // post-loop bookkeeping below keeps the - // `last_seen_id` field stable. - newest_id = Some(first.to_string()); - break; - } - } - } - - // ── Step 5: filter against synced_ids for early-stop, advance - // cursor tracker, and collect new messages for batched - // memory-tree ingest. We collect candidate IDs to mark - // synced but defer the mark until the batch ingest returns - // Ok — otherwise a total ingest failure would leave these - // messages flagged as synced (gmail-side fetch dedup) but - // NOT in the memory tree, with no way to retry. - let mut all_already_synced = true; - let mut new_messages: Vec = Vec::with_capacity(messages.len()); - let mut pending_synced_ids: Vec = Vec::with_capacity(messages.len()); - for (msg_index, msg) in messages.iter().enumerate() { - // Track the newest date we've seen for cursor advancement, - // independent of dedup status — we want the cursor to move - // even if we've already ingested this page's content. - if let Some(date_val) = extract_item_id(msg, MESSAGE_DATE_PATHS) { - if newest_date - .as_ref() - .is_none_or(|existing| date_val > *existing) - { - newest_date = Some(date_val); - } - } - - let msg_id = extract_item_id(msg, MESSAGE_ID_PATHS); - // Capture the very first id of page 0 as the - // freshest-id-on-server marker for next-sync's - // head-unchanged shortcut, regardless of dedup status. - if page_num == 0 && msg_index == 0 { - if let Some(ref id) = msg_id { - newest_id = Some(id.clone()); - } - } - if let Some(ref id) = msg_id { - if state.is_synced(id) { - continue; - } - pending_synced_ids.push(id.clone()); - } - all_already_synced = false; - new_messages.push(msg.clone()); - } - - // ctx.max_items precise cap: clamp the per-page batch before ingest - // so a single page larger than the budget is never over-persisted. - cap.clamp_batch(&mut new_messages); - cap.clamp_batch(&mut pending_synced_ids); - - // Single batched ingest into memory_tree. Chunk IDs are - // content-hashed so re-ingest of the same message is an - // idempotent UPSERT at the SQL layer; per-message dedup above - // is purely an optimisation for the hot path. - // - // `synced_ids` here means "Gmail-side fetch dedup" (don't burn - // API quota re-fetching this message), not "fully durable in - // memory tree". We only commit those marks once the batch - // returns Ok; on Err, nothing is marked, so the next sync - // re-fetches and the chunk-id content hash handles dedup at - // the storage layer. - if !new_messages.is_empty() { - let owner = format!("gmail-sync:{connection_id}"); - match ingest_page_into_memory_tree_with_outcome( - ctx.config.as_ref(), - &owner, - account_email.as_deref(), - &new_messages, - ) - .await - { - 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 += outcome.item_ids_ingested.len(); - cap.record(outcome.item_ids_ingested.len()); - tracing::debug!( - page = page_num, - new_messages = new_messages.len(), - 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, - new_messages = new_messages.len(), - "[composio:gmail] ingest_page_into_memory_tree failed (continuing)" - ); - } - } - } - - // If every message in this page was already synced, there's - // nothing new beyond this point — stop paginating. - if all_already_synced { - tracing::debug!( - page = page_num, - "[composio:gmail] all items in page already synced, stopping" - ); - stop_reason = "page_all_synced"; - break; - } - - // ctx.max_items hard stop: break once the per-source cap is reached. - if cap.is_reached() { - tracing::debug!( - page = page_num, - total_persisted, - "[composio:gmail] [memory_sync] max_items reached, stopping pagination" - ); - stop_reason = "max_items"; - hit_cap_boundary = true; - break; - } - - // Check for next page token. - page_token = sync::extract_page_token(&resp.data); - if page_token.is_none() { - tracing::debug!(page = page_num, "[composio:gmail] no next page token, done"); - stop_reason = "no_more_pages"; - break; - } - } - - // ── 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 !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, - 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 { - state.set_last_seen_id(freshest); - } - let finished_at_ms = sync::now_ms(); - state.set_last_sync_at_ms(finished_at_ms); - state.save(&memory).await?; - - // Bump the in-process scheduler timestamp so a periodic tick - // does not immediately re-fire on top of a trigger-driven or - // connection-created sync. Periodic itself already calls this - // on its own success path; calling it from the provider keeps - // the bookkeeping consistent for the other entry points. - crate::openhuman::memory_sync::composio::periodic::record_sync_success( - self.toolkit_slug(), - &connection_id, - ); - - let dup_ratio = if total_fetched > 0 { - (total_fetched.saturating_sub(total_persisted)) as f64 / total_fetched as f64 - } else { - 0.0 - }; - let summary = format!( - "gmail sync ({reason}): fetched {total_fetched}, persisted {total_persisted} new, \ - requests {total_requests}, budget remaining {remaining}, stop={stop}", - reason = reason.as_str(), - remaining = state.budget_remaining(), - stop = stop_reason, - ); - tracing::info!( - connection_id = %connection_id, - reason = reason.as_str(), - elapsed_ms = finished_at_ms.saturating_sub(started_at_ms), - requests = total_requests, - messages_total = total_fetched, - messages_new = total_persisted, - dup_ratio = dup_ratio, - stop_reason = stop_reason, - budget_remaining = state.budget_remaining(), - adaptive_cap = max_pages != MAX_PAGES_PER_SYNC, - "[composio:gmail] incremental sync complete" - ); - - Ok(SyncOutcome { - toolkit: "gmail".to_string(), - connection_id: Some(connection_id), - reason: reason.as_str().to_string(), - items_ingested: total_persisted, - started_at_ms, - finished_at_ms, - summary, - details: json!({ - "messages_fetched": total_fetched, - "messages_persisted": total_persisted, - "requests": total_requests, - "budget_remaining": state.budget_remaining(), - "cursor": state.cursor, - "last_seen_id": state.last_seen_id, - "stop_reason": stop_reason, - "adaptive_cap": max_pages != MAX_PAGES_PER_SYNC, - "dup_ratio": dup_ratio, - "synced_ids_total": state.synced_ids.len(), - }), - }) + run_gmail_sync(ctx, reason).await } async fn on_trigger( diff --git a/src/openhuman/memory_sync/composio/providers/gmail/source.rs b/src/openhuman/memory_sync/composio/providers/gmail/source.rs new file mode 100644 index 000000000..2219b423e --- /dev/null +++ b/src/openhuman/memory_sync/composio/providers/gmail/source.rs @@ -0,0 +1,363 @@ +//! Gmail's [`IncrementalSource`] primitives. +//! +//! Gmail rides the generic +//! [`crate::openhuman::memory_sync::composio::providers::orchestrator`]: +//! [`GmailProvider::sync`](super::provider::GmailProvider) delegates to +//! [`run_gmail_sync`]. The orchestrator owns the control flow (budget, +//! pagination bound, dedup, the precise `max_items` clamp, cursor advance/hold, +//! state persistence); this module supplies only the Gmail-specific shapes. +//! +//! Gmail is **flat**, with three quirks handled by trait hooks: +//! * **server-side depth** — the `sync_depth_days` window is injected as an +//! `after:` query qualifier on the first sync (no cursor), so it +//! overrides [`IncrementalSource::server_side_depth`]. +//! * **adaptive page ceiling** — when the previous sync ran within the last +//! few minutes, [`GmailSource::page_ceiling`] caps pagination aggressively. +//! * **stop on an all-synced page** — Gmail's result set is newest-first, so a +//! page whose messages are all already-synced means nothing newer is left +//! ([`IncrementalSource::stop_on_empty_pending`] = `true`). +//! +//! The account email (used as the per-account ingest `source_id`) is resolved +//! once in the preamble — **not** budget-counted, matching the original — and +//! stashed on the source so [`GmailSource::ingest`] can read it back. Messages +//! ingest as a single batch per page; dedup is keyed by the bare message id +//! (emails are immutable). +//! +//! ## One intentional behavior drop +//! +//! The legacy loop also had a `last_seen_id` "head-unchanged" micro-optimization +//! (stop before post-processing page 0 when its first id matches the last sync's +//! freshest id). That is **redundant** with `stop_on_empty_pending`: on a quiet +//! inbox both fetch exactly one page and stop, so the API-budget behavior is +//! unchanged. It is dropped here (and `last_seen_id` is no longer written) to +//! avoid a gmail-only orchestrator hook for a no-op-on-budget optimization. + +use std::sync::OnceLock; + +use async_trait::async_trait; +use serde_json::{json, Value}; + +use super::ingest::ingest_page_into_memory_tree_with_outcome; +use super::provider::{ACTION_FETCH_EMAILS, ACTION_GET_PROFILE, BASE_QUERY}; +use super::sync; +use crate::openhuman::memory_sync::composio::providers::orchestrator::{ + self, IncrementalSource, IngestOutcome, PageFetch, SyncItem, SyncScope, +}; +use crate::openhuman::memory_sync::composio::providers::sync_state::{extract_item_id, SyncState}; +use crate::openhuman::memory_sync::composio::providers::{ + pick_str, ProviderContext, SyncOutcome, SyncReason, +}; + +/// Page size per API call. +const PAGE_SIZE: u32 = 25; + +/// Larger page size for the very first sync after OAuth. +const INITIAL_PAGE_SIZE: u32 = 50; + +/// Maximum pages to fetch in a single sync pass. +const MAX_PAGES_PER_SYNC: u32 = 20; + +/// Adaptive page cap applied when a successful sync ran very recently. +const RECENT_SYNC_MAX_PAGES: u32 = 2; + +/// "Recent" window (ms) used by the adaptive page cap. +const RECENT_SYNC_WINDOW_MS: u64 = 5 * 60 * 1000; + +/// Paths to try when extracting a message's unique id. +const MESSAGE_ID_PATHS: &[&str] = &["id", "data.id", "messageId", "data.messageId"]; + +/// Paths for extracting the internal date (epoch millis or date string) used as +/// the sync cursor. +const MESSAGE_DATE_PATHS: &[&str] = &[ + "internalDate", + "data.internalDate", + "date", + "data.date", + "receivedAt", + "data.receivedAt", +]; + +/// Gmail's [`IncrementalSource`]. Holds the account email resolved in the +/// preamble so [`Self::ingest`] can stamp a stable per-account `source_id`. +pub(crate) struct GmailSource { + account_email: OnceLock>, +} + +impl GmailSource { + fn new() -> Self { + Self { + account_email: OnceLock::new(), + } + } + + /// Resolve the account email via `GMAIL_GET_PROFILE`. Tolerant of failure + /// (returns `None` → ingest falls back to per-participants bucketing) and + /// **not** budget-counted, matching the original sync. + async fn resolve_account_email(&self, ctx: &ProviderContext) -> Option { + match ctx.execute(ACTION_GET_PROFILE, Some(json!({}))).await { + Ok(resp) if resp.successful => pick_str( + &resp.data, + &["emailAddress", "email", "profile.emailAddress"], + ), + Ok(_) | Err(_) => None, + } + } +} + +/// Entry point used by [`super::provider::GmailProvider::sync`]. Bumps the +/// in-process scheduler heartbeat on completion (parity with the original), +/// so a periodic tick doesn't immediately re-fire on top of a trigger- or +/// connection-created sync. +pub(crate) async fn run_gmail_sync( + ctx: &ProviderContext, + reason: SyncReason, +) -> Result { + let connection_id = ctx + .connection_id + .clone() + .unwrap_or_else(|| "default".to_string()); + let outcome = orchestrator::run_sync(&GmailSource::new(), ctx, reason).await?; + crate::openhuman::memory_sync::composio::periodic::record_sync_success("gmail", &connection_id); + Ok(outcome) +} + +#[async_trait] +impl IncrementalSource for GmailSource { + fn toolkit(&self) -> &'static str { + "gmail" + } + + fn page_size(&self, reason: SyncReason) -> u32 { + match reason { + SyncReason::ConnectionCreated => INITIAL_PAGE_SIZE, + _ => PAGE_SIZE, + } + } + + fn max_pages(&self) -> u32 { + MAX_PAGES_PER_SYNC + } + + /// Adaptive cap: an initial backfill gets the full ceiling; a steady-state + /// sync that ran within [`RECENT_SYNC_WINDOW_MS`] is capped aggressively + /// (recent ticks rarely need more than a couple of pages). + fn page_ceiling(&self, reason: SyncReason, state: &SyncState) -> u32 { + match reason { + SyncReason::ConnectionCreated => MAX_PAGES_PER_SYNC, + _ => match state.last_sync_at_ms { + Some(last_ms) if sync::now_ms().saturating_sub(last_ms) < RECENT_SYNC_WINDOW_MS => { + RECENT_SYNC_MAX_PAGES + } + _ => MAX_PAGES_PER_SYNC, + }, + } + } + + fn stop_on_empty_pending(&self) -> bool { + true + } + + fn server_side_depth(&self) -> bool { + true + } + + fn detail_noun(&self) -> &'static str { + "messages" + } + + /// Resolve the account email (stashed for `ingest`) and return the single + /// flat scope. The profile fetch is not budget-counted (parity). + async fn preamble( + &self, + ctx: &ProviderContext, + _state: &mut SyncState, + ) -> Result, String> { + let email = self.resolve_account_email(ctx).await; + if email.is_none() { + tracing::warn!( + connection_id = ?ctx.connection_id, + "[composio:gmail] account email unresolved; ingest falls back to per-participants source_id" + ); + } + let _ = self.account_email.set(email); + Ok(vec![SyncScope::flat()]) + } + + async fn fetch_page( + &self, + ctx: &ProviderContext, + _scope: &SyncScope, + cursor: Option<&str>, + reason: SyncReason, + state: &mut SyncState, + ) -> Result { + // Build the Gmail search query. Prefer a second-precision + // `after:` from the persistent cursor; on the first sync inject + // the `sync_depth_days` floor (server-side depth). + let mut query = BASE_QUERY.to_string(); + if let Some(persistent) = state.cursor.as_deref() { + if let Some(epoch_filter) = sync::cursor_to_gmail_after_epoch_filter(persistent) { + query.push_str(&format!(" after:{epoch_filter}")); + } else if let Some(date_filter) = sync::cursor_to_gmail_after_filter(persistent) { + query.push_str(&format!(" after:{date_filter}")); + } + } else if let Some(days) = ctx.sync_depth_days { + let floor_secs = super::super::helpers::epoch_floor_from_depth(days); + query.push_str(&format!(" after:{floor_secs}")); + } + + let mut args = json!({ + "max_results": self.page_size(reason), + "query": query, + }); + // The orchestrator's opaque cursor carries Gmail's page token. + if let Some(token) = cursor { + args["page_token"] = json!(token); + } + + let mut resp = ctx + .execute(ACTION_FETCH_EMAILS, Some(args.clone())) + .await + .map_err(|e| format!("[composio:gmail] {ACTION_FETCH_EMAILS}: {e:#}"))?; + state.record_requests(1); + + if !resp.successful { + let err = resp + .error + .clone() + .unwrap_or_else(|| "provider reported failure".to_string()); + return Err(format!("[composio:gmail] {ACTION_FETCH_EMAILS}: {err}")); + } + + // Pull the backend's pre-rendered `markdownFormatted` onto each message + // before post-process, then slim/normalize the envelope (same order as + // the original sync). + if let Some(top_md) = resp + .markdown_formatted + .as_deref() + .map(str::trim) + .filter(|s| !s.is_empty()) + { + super::post_process::apply_response_level_markdown(&mut resp.data, top_md); + } + super::post_process::post_process(ACTION_FETCH_EMAILS, Some(&args), &mut resp.data); + + Ok(PageFetch { + items: sync::extract_messages(&resp.data), + next: sync::extract_page_token(&resp.data), + }) + } + + /// Emails are immutable — dedup by the bare message id (no `@version`). + fn item_dedup_key(&self, item: &Value) -> Option { + extract_item_id(item, MESSAGE_ID_PATHS) + } + + fn item_sort_ts(&self, item: &Value) -> Option { + extract_item_id(item, MESSAGE_DATE_PATHS) + } + + /// Batch-ingest the page's new messages into the memory tree. `synced_keys` + /// are the message ids that actually ingested; a partial ingest trips + /// `had_failures` so the orchestrator holds the cursor for retry. + async fn ingest( + &self, + ctx: &ProviderContext, + _scope: &SyncScope, + _state: &mut SyncState, + items: Vec, + ) -> IngestOutcome { + if items.is_empty() { + return IngestOutcome::default(); + } + let connection_id = ctx.connection_id.as_deref().unwrap_or("default"); + let owner = format!("gmail-sync:{connection_id}"); + let account_email = self.account_email.get().cloned().flatten(); + let total = items.len(); + let messages: Vec = items.into_iter().map(|it| it.raw).collect(); + + match ingest_page_into_memory_tree_with_outcome( + ctx.config.as_ref(), + &owner, + account_email.as_deref(), + &messages, + ) + .await + { + Ok(outcome) => { + let persisted = outcome.item_ids_ingested.len(); + IngestOutcome { + synced_keys: outcome.item_ids_ingested, + persisted, + // A short ingest (fewer messages persisted than handed in) + // holds the cursor so the next sync re-scans the gap. + had_failures: persisted < total, + } + } + Err(e) => { + tracing::warn!( + error = %format!("{e:#}"), + new_messages = total, + "[composio:gmail] ingest_page_into_memory_tree failed (continuing)" + ); + IngestOutcome { + synced_keys: Vec::new(), + persisted: 0, + had_failures: true, + } + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + #[test] + fn item_dedup_key_is_bare_message_id() { + // No `@version` — emails are immutable. + assert_eq!( + GmailSource::new() + .item_dedup_key(&json!({ "messageId": "m1", "internalDate": "1780000000000" })) + .as_deref(), + Some("m1") + ); + assert_eq!( + GmailSource::new().item_dedup_key(&json!({ "internalDate": "x" })), + None + ); + } + + #[test] + fn gmail_advertises_server_side_depth_and_empty_stop() { + let s = GmailSource::new(); + assert!(s.server_side_depth()); + assert!(s.stop_on_empty_pending()); + assert_eq!(s.detail_noun(), "messages"); + } + + #[test] + fn page_ceiling_caps_aggressively_after_a_recent_sync() { + let s = GmailSource::new(); + let mut state = SyncState::new("gmail", "c"); + // Initial backfill always gets the full ceiling. + assert_eq!( + s.page_ceiling(SyncReason::ConnectionCreated, &state), + MAX_PAGES_PER_SYNC + ); + // A steady-state sync right after the previous one is capped. + state.set_last_sync_at_ms(sync::now_ms()); + assert_eq!( + s.page_ceiling(SyncReason::Periodic, &state), + RECENT_SYNC_MAX_PAGES + ); + // A stale last-sync drops back to the full ceiling. + state.set_last_sync_at_ms(0); + assert_eq!( + s.page_ceiling(SyncReason::Periodic, &state), + MAX_PAGES_PER_SYNC + ); + } +} diff --git a/src/openhuman/memory_sync/composio/providers/orchestrator.rs b/src/openhuman/memory_sync/composio/providers/orchestrator.rs index 79521493b..312e31b36 100644 --- a/src/openhuman/memory_sync/composio/providers/orchestrator.rs +++ b/src/openhuman/memory_sync/composio/providers/orchestrator.rs @@ -135,6 +135,24 @@ pub(crate) trait IncrementalSource: Send + Sync { /// [`ItemCap::max_pages`], applied *per scope*. fn max_pages(&self) -> u32; + /// The page ceiling for *this pass*, given the reason and current state. + /// Defaults to the fixed [`Self::max_pages`]; gmail overrides it to apply an + /// adaptive cap (e.g. only a couple of pages when the previous sync ran very + /// recently). The orchestrator hands the result to [`ItemCap::max_pages`]. + fn page_ceiling(&self, reason: SyncReason, state: &SyncState) -> u32 { + let _ = (reason, state); + self.max_pages() + } + + /// Whether to stop paginating a scope once a fetched page yields **no new + /// items** (everything deduped away as already-synced). Default `false` — + /// the orchestrator keeps paginating until the cursor/depth boundary or the + /// last page. Gmail overrides to `true`: its result set is ordered so an + /// all-already-synced page means there is nothing newer left to fetch. + fn stop_on_empty_pending(&self) -> bool { + false + } + /// Resolve identity and list the scopes to iterate. /// /// Flat providers return `vec![SyncScope::flat()]`. Nested providers call @@ -435,8 +453,11 @@ pub(crate) async fn run_sync( // ── Step 4: caps + window ─────────────────────────────────────────── let page_size = source.page_size(reason); let mut cap = ItemCap::new(ctx.max_items); - let effective_max_pages = cap.max_pages(page_size, source.max_pages()); - if ctx.max_items.is_some() && effective_max_pages < source.max_pages() { + // `page_ceiling` lets a provider apply a state-aware cap (gmail's adaptive + // cap); it defaults to the fixed `max_pages`. + let page_ceiling = source.page_ceiling(reason, &state); + let effective_max_pages = cap.max_pages(page_size, page_ceiling); + if ctx.max_items.is_some() && effective_max_pages < page_ceiling { tracing::debug!( toolkit, connection_id = %connection_id, @@ -560,6 +581,11 @@ pub(crate) async fn run_sync( let (mut pending, mut hit_cursor_boundary) = select_pending(source, &fetch.items, boundary_cursor, &state, ts_acc); + // A non-empty page whose items all deduped away as already-synced + // (the gmail "all already synced" signal — captured before the + // depth/cap filters narrow `pending` for other reasons). + let page_had_no_new_items = pending.is_empty(); + // sync_depth_days: `pending` is in descending-timestamp order, so // truncate at the first item below the floor and stop paginating. if let Some(ref floor) = depth_floor { @@ -608,6 +634,16 @@ pub(crate) async fn run_sync( break; } + if page_had_no_new_items && source.stop_on_empty_pending() { + tracing::debug!( + toolkit, + scope = %scope.label, + page = page_num, + "[composio:sync_orch] page had no new items, stopping scope" + ); + break; + } + cursor = fetch.next; if cursor.is_none() { tracing::debug!( @@ -780,6 +816,14 @@ mod tests { /// Scope id whose `fetch_page` returns a transport error (used to /// exercise per-scope error tolerance). `None` → every scope succeeds. fail_scope: Option, + /// When `Some`, drive multi-page returns: page `i` returns `pages[i]` + /// with the opaque cursor = next page index. Overrides the single-page + /// synth path. + pages: Option>>, + /// When true, override `stop_on_empty_pending()` (gmail's behaviour). + stop_on_empty: bool, + /// When `Some`, override `page_ceiling()` (gmail's adaptive cap). + page_ceiling: Option, } impl FakeSource { @@ -803,6 +847,12 @@ mod tests { fn max_pages(&self) -> u32 { 20 } + fn stop_on_empty_pending(&self) -> bool { + self.stop_on_empty + } + fn page_ceiling(&self, _reason: SyncReason, _state: &SyncState) -> u32 { + self.page_ceiling.unwrap_or_else(|| self.max_pages()) + } async fn preamble( &self, _ctx: &ProviderContext, @@ -855,6 +905,13 @@ mod tests { // Completed but the provider reported failure — already recorded. return Err("fake provider-reported page failure".to_string()); } + // Multi-page mode: cursor is the page index. + if let Some(pages) = &self.pages { + let idx: usize = cursor.and_then(|c| c.parse().ok()).unwrap_or(0); + let items = pages.get(idx).cloned().unwrap_or_default(); + let next = (idx + 1 < pages.len()).then(|| (idx + 1).to_string()); + return Ok(PageFetch { items, next }); + } // Single page per scope: everything comes back on the first call. if cursor.is_some() { return Ok(PageFetch { @@ -1006,6 +1063,75 @@ mod tests { ); } + /// Two pages of items, keyed `id`+`ts`. + fn page(prefix: &str, n: usize) -> Vec { + (0..n) + .map(|i| json!({ "id": format!("{prefix}{i}"), "ts": "2099-01-01T00:00:00Z" })) + .collect() + } + + #[tokio::test] + async fn page_ceiling_caps_the_pages_fetched() { + let tmp = TempDir::new().unwrap(); + let ctx = fake_ctx(&tmp, None, None); + // Two pages of 2 items each. With a page ceiling of 1 (gmail's adaptive + // cap), only the first page is fetched → 2 ingested, not 4. + let source = FakeSource { + scopes: vec![SyncScope::flat()], + pages: Some(vec![page("a", 2), page("b", 2)]), + page_ceiling: Some(1), + ..Default::default() + }; + let outcome = run_sync(&source, &ctx, SyncReason::Periodic) + .await + .expect("run_sync"); + assert_eq!( + outcome.items_ingested, 2, + "page_ceiling=1 must fetch only the first page" + ); + } + + #[tokio::test] + async fn stop_on_empty_pending_halts_at_an_all_synced_page() { + let tmp = TempDir::new().unwrap(); + // Page 0 = [a0,a1] (new), page 1 = same ids (all synced after page 0), + // page 2 = [c0,c1] (new). With stop_on_empty the pass halts at page 1 + // and never reaches page 2 → 2 ingested; without it, page 2's items are + // also ingested → 4. + let pages = vec![page("a", 2), page("a", 2), page("c", 2)]; + + let stop = FakeSource { + scopes: vec![SyncScope::flat()], + pages: Some(pages.clone()), + stop_on_empty: true, + ..Default::default() + }; + let stop_ctx = fake_ctx(&tmp, None, None); + let stopped = run_sync(&stop, &stop_ctx, SyncReason::Periodic) + .await + .expect("run_sync"); + assert_eq!( + stopped.items_ingested, 2, + "stop_on_empty must halt at the all-synced page before page 2" + ); + + // Control: same pages, no stop_on_empty → keeps going to page 2. + let tmp2 = TempDir::new().unwrap(); + let keep = FakeSource { + scopes: vec![SyncScope::flat()], + pages: Some(pages), + ..Default::default() + }; + let keep_ctx = fake_ctx(&tmp2, None, None); + let kept = run_sync(&keep, &keep_ctx, SyncReason::Periodic) + .await + .expect("run_sync"); + assert_eq!( + kept.items_ingested, 4, + "without stop_on_empty the pass continues past the all-synced page" + ); + } + #[tokio::test] async fn nested_scopes_share_one_cap_budget() { let tmp = TempDir::new().unwrap(); diff --git a/tests/memory_sync_providers_raw_coverage_e2e.rs b/tests/memory_sync_providers_raw_coverage_e2e.rs index 38c95e095..952b72887 100644 --- a/tests/memory_sync_providers_raw_coverage_e2e.rs +++ b/tests/memory_sync_providers_raw_coverage_e2e.rs @@ -41,6 +41,7 @@ use openhuman_core::openhuman::memory_sync::composio::providers::slack::ingest a use openhuman_core::openhuman::memory_sync::composio::providers::slack::{ SlackMessage, SlackProvider, }; +use openhuman_core::openhuman::memory_sync::composio::providers::sync_state::SyncState; use openhuman_core::openhuman::memory_sync::composio::providers::{ ComposioProvider, ProviderContext, SyncReason, TaskFetchFilter, }; @@ -992,6 +993,161 @@ async fn gmail_sync_max_items_caps_ingest_to_exact_count() { server.abort(); } +#[tokio::test] +async fn gmail_sync_depth_days_injects_after_floor_into_query() { + 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"); + + // Gmail applies sync_depth_days server-side: on the first sync (no cursor) + // it must inject an `after:` qualifier into the GMAIL_FETCH_EMAILS + // query, proving the window actually narrows what is fetched. + let requests: Arc>> = Arc::new(Mutex::new(Vec::new())); + let (base, server) = loopback_router(gmail_cap_router( + gmail_cap_messages(1), + Arc::clone(&requests), + )) + .await; + + let mut config = config_in(&tmp); + config.api_url = Some(base); + persist_config(&config).await; + store_session(&config); + memory_global::init(config.workspace_dir.clone()).expect("init global memory"); + + let ctx = ProviderContext { + config: Arc::new(config), + toolkit: "gmail".to_string(), + connection_id: Some("conn-gmail-depth".to_string()), + usage: Default::default(), + max_items: None, + sync_depth_days: Some(7), + }; + + GmailProvider::new() + .sync(&ctx, SyncReason::ConnectionCreated) + .await + .expect("gmail depth sync"); + + let reqs = requests.lock().unwrap(); + let fetch = reqs + .iter() + .find(|b| b.get("tool").and_then(Value::as_str) == Some("GMAIL_FETCH_EMAILS")) + .expect("a GMAIL_FETCH_EMAILS request was issued"); + let query = fetch + .get("arguments") + .and_then(|a| a.get("query")) + .and_then(Value::as_str) + .unwrap_or(""); + assert!( + query.contains("after:"), + "sync_depth_days=7 must inject an `after:` floor into the query, got: {query}" + ); + + server.abort(); +} + +/// Router whose `GMAIL_FETCH_EMAILS` always returns the same page **with a +/// non-empty `nextPageToken`** — so absent an all-synced early stop the provider +/// would paginate to the page ceiling. Used to pin `stop_on_empty_pending`. +fn gmail_repeating_router(messages: Vec, requests: Arc>>) -> Router { + Router::new().route( + "/agent-integrations/composio/execute", + any(move |Json(body): Json| { + let messages = messages.clone(); + let requests = Arc::clone(&requests); + async move { + requests.lock().unwrap().push(body.clone()); + let tool = body.get("tool").and_then(Value::as_str).unwrap_or(""); + let resp = match tool { + "GMAIL_GET_PROFILE" => { + execute_envelope(json!({ "emailAddress": "stop@example.test" })) + } + "GMAIL_FETCH_EMAILS" => execute_envelope(json!({ + "messages": messages, + "nextPageToken": "more" + })), + _ => execute_envelope(json!({})), + }; + Json(resp) + } + }), + ) +} + +// Replaces gmail's dropped `last_seen_id` head-unchanged optimization: prove the +// provider halts after a single `GMAIL_FETCH_EMAILS` round-trip when the first +// page is already fully synced. We pre-seed `synced_ids` with the page's message +// ids and leave the cursor `None` — so the cursor/depth boundary cannot fire and +// **only** `stop_on_empty_pending` can stop the loop. The router hands back a +// perpetual `nextPageToken`, so without the early stop the provider would walk +// every page up to the ceiling. +#[tokio::test] +async fn gmail_sync_stops_after_an_all_already_synced_page() { + 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 requests: Arc>> = Arc::new(Mutex::new(Vec::new())); + let (base, server) = loopback_router(gmail_repeating_router( + gmail_cap_messages(3), + Arc::clone(&requests), + )) + .await; + + let mut config = config_in(&tmp); + config.api_url = Some(base); + persist_config(&config).await; + store_session(&config); + let memory = memory_global::init(config.workspace_dir.clone()).expect("init global memory"); + + // Pre-seed: the 3 page-1 message ids are already synced; cursor stays None. + let mut state = SyncState::new("gmail", "conn-gmail-stop"); + for i in 1..=3 { + state.mark_synced(format!("gmail-cap-msg-{i}")); + } + state + .save(&memory) + .await + .expect("save pre-seeded sync state"); + + let ctx = ProviderContext { + config: Arc::new(config), + toolkit: "gmail".to_string(), + connection_id: Some("conn-gmail-stop".to_string()), + usage: Default::default(), + max_items: None, + sync_depth_days: None, + }; + + let outcome = GmailProvider::new() + .sync(&ctx, SyncReason::ConnectionCreated) + .await + .expect("gmail sync"); + + assert_eq!( + outcome.items_ingested, 0, + "every page-1 message was already synced — nothing to ingest" + ); + let fetch_calls = requests + .lock() + .unwrap() + .iter() + .filter(|b| b.get("tool").and_then(Value::as_str) == Some("GMAIL_FETCH_EMAILS")) + .count(); + assert_eq!( + fetch_calls, 1, + "stop_on_empty_pending must halt after one all-synced page, not paginate the \ + perpetual nextPageToken — got {fetch_calls} GMAIL_FETCH_EMAILS calls" + ); + + server.abort(); +} + // ───────────────────────────────────────────────────────────────────────── // Notion cap enforcement //