perf(composio/gmail): cut redundant fetches on incremental sync (#1404) (#1474)

Co-authored-by: Steven Enamakel <enamakel@tinyhumans.ai>

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<String> = None;
+        let mut newest_id: Option<String> = None;
         let mut page_token: Option<String> = 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:<unix>` 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<Value> = Vec::with_capacity(messages.len());
             let mut pending_synced_ids: Vec<String> = 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(),
             }),
         })

src/openhuman/composio/providers/gmail/sync.rs

@@ -39,8 +39,12 @@ pub(crate) fn extract_page_token(data: &Value) -> Option<String> {
 }
 
 /// 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:<unix>` filter and so avoids re-fetching
+/// large same-day windows on every tick.
 pub(crate) fn cursor_to_gmail_after_filter(cursor: &str) -> Option<String> {
     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<String> {
     None
 }
 
+/// Convert a cursor value (epoch millis or date string) into a Gmail
+/// `after:<unix_seconds>` 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<String> {
+    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<i64> {
+    let cursor = cursor.trim();
+    if let Ok(millis) = cursor.parse::<i64>() {
+        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}`");
+        }
+    }
 }

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

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<String>,
+
+    /// 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<u64>,
 }
 
 /// 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<String>) {
+        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]