diff --git a/src/openhuman/composio/providers/gmail/provider.rs b/src/openhuman/composio/providers/gmail/provider.rs index 01b032da2..fd1bb2758 100644 --- a/src/openhuman/composio/providers/gmail/provider.rs +++ b/src/openhuman/composio/providers/gmail/provider.rs @@ -49,6 +49,20 @@ const INITIAL_PAGE_SIZE: u32 = 50; /// 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"]; @@ -225,31 +239,66 @@ impl ComposioProvider for GmailProvider { _ => 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 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, + }, + }; + 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"; - for page_num in 0..MAX_PAGES_PER_SYNC { + 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. If we have a cursor (date of last - // synced message), add `after:YYYY/MM/DD` so the API only - // returns newer mail. + // 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. let mut query = "in:inbox -in:spam -in:trash".to_string(); if let Some(ref cursor) = state.cursor { - if let Some(date_filter) = sync::cursor_to_gmail_after_filter(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 date filter from cursor" + "[composio:gmail] using day-level filter from cursor (epoch parse failed)" ); } } @@ -271,6 +320,7 @@ impl ComposioProvider for GmailProvider { })?; state.record_requests(1); + total_requests += 1; if !resp.successful { let err = resp @@ -309,9 +359,39 @@ impl ComposioProvider for GmailProvider { 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 @@ -322,7 +402,7 @@ impl ComposioProvider for GmailProvider { 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 in &messages { + 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. @@ -336,6 +416,14 @@ impl ComposioProvider for GmailProvider { } 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; @@ -401,6 +489,7 @@ impl ComposioProvider for GmailProvider { page = page_num, "[composio:gmail] all items in page already synced, stopping" ); + stop_reason = "page_all_synced"; break; } @@ -408,6 +497,7 @@ impl ComposioProvider for GmailProvider { 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; } } @@ -416,21 +506,46 @@ impl ComposioProvider for GmailProvider { if let Some(new_cursor) = newest_date { state.advance_cursor(&new_cursor); } + 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?; - let finished_at_ms = sync::now_ms(); + // 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::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, \ - budget remaining {remaining}", + 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), - total_fetched, - total_persisted, + 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" ); @@ -445,8 +560,13 @@ impl ComposioProvider for GmailProvider { 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(), }), }) diff --git a/src/openhuman/composio/providers/gmail/sync.rs b/src/openhuman/composio/providers/gmail/sync.rs index eb2652820..2860aeb48 100644 --- a/src/openhuman/composio/providers/gmail/sync.rs +++ b/src/openhuman/composio/providers/gmail/sync.rs @@ -39,8 +39,12 @@ pub(crate) fn extract_page_token(data: &Value) -> Option { } /// Convert a cursor value (epoch millis or date string) into a Gmail -/// `after:YYYY/MM/DD` filter component. Returns `None` if the cursor -/// cannot be parsed. +/// `after:YYYY/MM/DD` filter component. Day-level precision — kept as a +/// last-resort filter for non-numeric cursors and for back-compat with +/// the older day-cursor write path. New code should prefer +/// [`cursor_to_gmail_after_epoch_filter`] which produces a +/// second-precision `after:` filter and so avoids re-fetching +/// large same-day windows on every tick. pub(crate) fn cursor_to_gmail_after_filter(cursor: &str) -> Option { let cursor = cursor.trim(); // Try parsing as epoch millis first (Gmail's internalDate). @@ -60,6 +64,39 @@ pub(crate) fn cursor_to_gmail_after_filter(cursor: &str) -> Option { None } +/// Convert a cursor value (epoch millis or date string) into a Gmail +/// `after:` filter component. Gmail's search syntax +/// accepts a bare unix-seconds value for `after:` / `before:`, so the +/// filter is second-precision rather than the day-level +/// `after:YYYY/MM/DD` form. Used by the incremental sync path so a +/// same-day re-tick does not re-fetch every message Gmail has filed +/// today. +/// +/// Returns `None` when the cursor cannot be parsed; callers should +/// fall back to the coarse day filter to avoid sending an unbounded +/// query. +pub(crate) fn cursor_to_gmail_after_epoch_filter(cursor: &str) -> Option { + let secs = parse_cursor_to_epoch_secs(cursor)?; + Some(secs.to_string()) +} + +/// Parse a cursor (epoch millis as string, `YYYY-MM-DD`, or RFC3339) +/// into unix-seconds. Shared by the epoch filter and by the adaptive +/// page-cap recency check. +pub(crate) fn parse_cursor_to_epoch_secs(cursor: &str) -> Option { + let cursor = cursor.trim(); + if let Ok(millis) = cursor.parse::() { + return Some(millis / 1000); + } + if let Ok(date) = chrono::NaiveDate::parse_from_str(cursor, "%Y-%m-%d") { + return date.and_hms_opt(0, 0, 0).map(|dt| dt.and_utc().timestamp()); + } + if let Ok(dt) = chrono::DateTime::parse_from_rfc3339(cursor) { + return Some(dt.timestamp()); + } + None +} + pub(crate) fn now_ms() -> u64 { use std::time::{SystemTime, UNIX_EPOCH}; SystemTime::now() @@ -158,4 +195,48 @@ mod tests { fn now_ms_returns_nonzero() { assert!(now_ms() > 0); } + + // ── second-precision cursor ────────────────────────────────── + + #[test] + fn epoch_filter_emits_unix_seconds_for_internal_date_millis() { + let filter = cursor_to_gmail_after_epoch_filter("1700000000000").unwrap(); + assert_eq!(filter, "1700000000"); + } + + #[test] + fn epoch_filter_handles_iso_date() { + let filter = cursor_to_gmail_after_epoch_filter("2024-01-15").unwrap(); + // 2024-01-15 00:00:00 UTC == 1705276800. + assert_eq!(filter, "1705276800"); + } + + #[test] + fn epoch_filter_handles_rfc3339() { + let filter = cursor_to_gmail_after_epoch_filter("2024-06-01T12:00:00Z").unwrap(); + assert_eq!(filter, "1717243200"); + } + + #[test] + fn epoch_filter_returns_none_for_garbage() { + assert!(cursor_to_gmail_after_epoch_filter("not-a-date").is_none()); + } + + #[test] + fn epoch_filter_trims_whitespace() { + let filter = cursor_to_gmail_after_epoch_filter(" 2024-01-15 ").unwrap(); + assert_eq!(filter, "1705276800"); + } + + #[test] + fn parse_cursor_round_trip_matches_epoch_filter() { + // The adaptive page-cap relies on parse_cursor_to_epoch_secs + // agreeing with cursor_to_gmail_after_epoch_filter — both must + // emit the same seconds value for any given input. + for cursor in ["1700000000000", "2024-01-15", "2024-06-01T12:00:00Z"] { + let secs = parse_cursor_to_epoch_secs(cursor).unwrap(); + let filter = cursor_to_gmail_after_epoch_filter(cursor).unwrap(); + assert_eq!(filter, secs.to_string(), "cursor `{cursor}`"); + } + } } diff --git a/src/openhuman/composio/providers/gmail/tests.rs b/src/openhuman/composio/providers/gmail/tests.rs index e9b93752e..4a6adb968 100644 --- a/src/openhuman/composio/providers/gmail/tests.rs +++ b/src/openhuman/composio/providers/gmail/tests.rs @@ -1,6 +1,9 @@ //! Unit tests for the Gmail provider. -use super::sync::{cursor_to_gmail_after_filter, extract_messages, extract_page_token}; +use super::sync::{ + cursor_to_gmail_after_epoch_filter, cursor_to_gmail_after_filter, extract_messages, + extract_page_token, +}; use super::GmailProvider; use crate::openhuman::composio::providers::ComposioProvider; use serde_json::json; @@ -81,6 +84,31 @@ fn default_impl_matches_new() { // Both are unit structs — constructing via Default is the cover target. } +#[test] +fn epoch_filter_is_preferred_over_day_filter_for_typical_internal_date() { + // The provider tries `cursor_to_gmail_after_epoch_filter` first + // and only falls back to the day filter if the parse fails. Both + // helpers must accept the same internalDate (epoch millis) input + // so the fallback path is genuinely a fallback, not a divergence. + let internal_date = "1774915200000"; // 2026-03-31 UTC + let epoch = cursor_to_gmail_after_epoch_filter(internal_date).unwrap(); + let day = cursor_to_gmail_after_filter(internal_date).unwrap(); + assert_eq!(epoch, "1774915200"); + assert_eq!(day, "2026/03/31"); + // Sanity bound: the epoch filter must be after 2020 and before + // year 2100, otherwise we shipped a regression in the cursor + // converter that would silently let queries land on year 1970. + let secs: i64 = epoch.parse().unwrap(); + assert!( + secs > 1_577_836_800, + "epoch filter must be after 2020-01-01" + ); + assert!( + secs < 4_102_444_800, + "epoch filter must be before 2100-01-01" + ); +} + // Note: full `sync` / `fetch_user_profile` / `on_trigger` paths require a // live `ComposioClient` (HTTP) plus the global `MemoryClient` singleton. // Those go through the integration test suite. Here we just lock in diff --git a/src/openhuman/composio/providers/sync_state.rs b/src/openhuman/composio/providers/sync_state.rs index 9c139d5ce..027fd0cf9 100644 --- a/src/openhuman/composio/providers/sync_state.rs +++ b/src/openhuman/composio/providers/sync_state.rs @@ -59,6 +59,31 @@ pub struct SyncState { /// Rolling daily request budget. #[serde(default)] pub daily_budget: DailyBudget, + + /// ID of the most recently synced item, used by providers (Gmail + /// today) to short-circuit a tick when the freshest server-side + /// item matches what we already have. Cheaper than re-walking the + /// `synced_ids` set — and crucially, it lets us bail out of + /// pagination on the very first page when nothing has changed, + /// instead of fetching `MAX_PAGES_PER_SYNC` worth of duplicates + /// before falling through the per-page dedup loop. + /// + /// `None` either when the state is fresh or when an older state + /// blob was loaded from disk that pre-dates this field. + #[serde(default)] + pub last_seen_id: Option, + + /// Unix milliseconds of the last successful sync that wrote into + /// memory. Lets the adaptive page-cap logic distinguish a "we + /// synced 30 seconds ago" tick (cap pages aggressively) from a + /// "we last synced two hours ago" tick (let pagination run to the + /// usual ceiling). Independent of the periodic scheduler's + /// in-process `LAST_SYNC_AT` map because that map is rebuilt on + /// every process restart whereas this value survives restarts. + /// + /// `None` until the first successful sync. + #[serde(default)] + pub last_sync_at_ms: Option, } /// Tracks the number of API requests made on a given calendar day. @@ -125,9 +150,24 @@ impl SyncState { cursor: None, synced_ids: HashSet::new(), daily_budget: DailyBudget::default(), + last_seen_id: None, + last_sync_at_ms: None, } } + /// Record the freshest item id observed on a successful sync. + /// Idempotent — repeated calls with the same id are no-ops. + pub fn set_last_seen_id(&mut self, item_id: impl Into) { + self.last_seen_id = Some(item_id.into()); + } + + /// Record the wall-clock time of a successful sync (unix + /// milliseconds). Persisted alongside the cursor so the adaptive + /// page-cap survives process restarts. + pub fn set_last_sync_at_ms(&mut self, ms: u64) { + self.last_sync_at_ms = Some(ms); + } + /// Whether the daily request budget is exhausted. pub fn budget_exhausted(&self) -> bool { self.daily_budget.is_exhausted() @@ -369,6 +409,8 @@ mod tests { state.mark_synced("item_a"); state.mark_synced("item_b"); state.daily_budget.record_requests(42); + state.set_last_seen_id("msg_top"); + state.set_last_sync_at_ms(1_700_000_000_000); let json = serde_json::to_value(&state).unwrap(); let restored: SyncState = serde_json::from_value(json).unwrap(); @@ -380,6 +422,44 @@ mod tests { assert!(restored.synced_ids.contains("item_b")); assert_eq!(restored.synced_ids.len(), 2); assert_eq!(restored.daily_budget.requests_used, 42); + assert_eq!(restored.last_seen_id.as_deref(), Some("msg_top")); + assert_eq!(restored.last_sync_at_ms, Some(1_700_000_000_000)); + } + + #[test] + fn sync_state_deserializes_legacy_blob_without_new_fields() { + // Older state blobs serialized before #1404 had no + // `last_seen_id` / `last_sync_at_ms` keys — make sure the + // deserializer still accepts them so existing users don't + // lose their cursor + dedup set on first upgrade. + let legacy = serde_json::json!({ + "toolkit": "gmail", + "connection_id": "conn_old", + "cursor": "1699000000000", + "synced_ids": ["m1", "m2"], + "daily_budget": { "date": today_str(), "requests_used": 7, "limit": 500 } + }); + let restored: SyncState = serde_json::from_value(legacy).unwrap(); + assert_eq!(restored.cursor.as_deref(), Some("1699000000000")); + assert_eq!(restored.synced_ids.len(), 2); + assert!(restored.last_seen_id.is_none()); + assert!(restored.last_sync_at_ms.is_none()); + } + + #[test] + fn set_last_seen_id_overwrites_previous_value() { + let mut state = SyncState::new("gmail", "c"); + state.set_last_seen_id("a"); + state.set_last_seen_id("b"); + assert_eq!(state.last_seen_id.as_deref(), Some("b")); + } + + #[test] + fn set_last_sync_at_ms_records_value() { + let mut state = SyncState::new("gmail", "c"); + state.set_last_sync_at_ms(123); + state.set_last_sync_at_ms(456); + assert_eq!(state.last_sync_at_ms, Some(456)); } #[test]