From 08178b5ff369845bcd1242da9dca64ba7dce4b9e Mon Sep 17 00:00:00 2001 From: Steven Enamakel <31011319+senamakel@users.noreply.github.com> Date: Sun, 12 Apr 2026 17:58:51 -0700 Subject: [PATCH] feat(composio): incremental sync with per-item persistence for Gmail and Notion (#519) * feat(composio): implement incremental sync and state management for Gmail and Notion providers - Enhanced the Gmail and Notion providers to support incremental synchronization with per-item persistence, improving data handling efficiency. - Introduced a new `SyncState` module to manage persistent sync state, including cursor tracking, synced IDs, and daily request budget management. - Updated sync logic to load state from a KV store, check daily budget limits, and handle paginated API requests, ensuring robust data retrieval and deduplication. - Refactored existing sync methods to utilize the new state management, enhancing overall reliability and performance of the providers. - Improved documentation for the sync process and state management, clarifying the operational flow and usage of the new features. * refactor(composio): improve error handling and logging in Gmail and Notion providers - Enhanced error handling in the Gmail provider to format error messages more clearly during email fetching. - Streamlined debug logging in both Gmail and Notion providers to improve readability by consolidating multiline statements into single lines. - Refactored the `extract_page_title` function in the Notion provider for better clarity in property extraction logic. - Overall, these changes aim to enhance maintainability and improve the clarity of error reporting and logging across the providers. --- src/openhuman/composio/providers/gmail.rs | 460 ++++++++++++------ src/openhuman/composio/providers/mod.rs | 1 + src/openhuman/composio/providers/notion.rs | 430 ++++++++++++---- .../composio/providers/sync_state.rs | 409 ++++++++++++++++ 4 files changed, 1058 insertions(+), 242 deletions(-) create mode 100644 src/openhuman/composio/providers/sync_state.rs diff --git a/src/openhuman/composio/providers/gmail.rs b/src/openhuman/composio/providers/gmail.rs index 6099697a6..a3ca661d8 100644 --- a/src/openhuman/composio/providers/gmail.rs +++ b/src/openhuman/composio/providers/gmail.rs @@ -1,42 +1,60 @@ -//! Gmail provider — native Rust counterpart to the QuickJS gmail skill. +//! Gmail provider — incremental sync with per-item persistence. //! -//! Mirrors the high-level shape of the JS skill in -//! `tinyhumansai/openhuman-skills/skills/gmail/index.js`: +//! On each sync pass: //! -//! * On connection / periodic tick → fetch the user profile -//! (`GMAIL_GET_PROFILE`) and a window of recent message metadata -//! (`GMAIL_FETCH_EMAILS`). -//! * Persist a JSON snapshot of the result into the global memory -//! layer under namespace `composio-gmail` so the agent loop can -//! surface it via `recall_memory`. -//! * On `GMAIL_NEW_GMAIL_MESSAGE` triggers → run an incremental -//! sync so newly arrived mail makes it into memory promptly. +//! 1. Load persistent [`SyncState`] from the KV store. +//! 2. Check the daily request budget — bail early if exhausted. +//! 3. Fetch a page of recent messages via `GMAIL_FETCH_EMAILS`, adding +//! a date filter when a cursor exists so only newer mail is returned. +//! 4. Deduplicate against `synced_ids` in the state. +//! 5. Persist each **new** message as its own memory document (not a +//! single giant snapshot) so agent recall can find individual emails. +//! 6. Paginate (up to budget) until no more results or all items in the +//! page are already synced. +//! 7. Advance the cursor and save state. //! -//! All upstream API access goes through -//! [`super::ProviderContext::client`] which proxies to the openhuman -//! backend's `/agent-integrations/composio/execute` endpoint. This -//! provider never holds raw OAuth tokens or hits Composio directly. +//! Daily budget (`DEFAULT_DAILY_REQUEST_LIMIT`, default 500) caps the +//! number of `execute_tool` calls per calendar day, preventing runaway +//! API usage during large initial backfills. use async_trait::async_trait; use serde_json::{json, Value}; +use super::sync_state::{extract_item_id, persist_single_item, SyncState}; use super::{ pick_str, ComposioProvider, ProviderContext, ProviderUserProfile, SyncOutcome, SyncReason, }; -/// Composio action slugs used by this provider. Hoisted to constants so -/// they're easy to grep + adjust if Composio renames them upstream. const ACTION_GET_PROFILE: &str = "GMAIL_GET_PROFILE"; const ACTION_FETCH_EMAILS: &str = "GMAIL_FETCH_EMAILS"; -/// Default page size for the periodic email pull. Kept conservative — -/// the goal is "freshness for the agent", not a full archive backfill. -const FETCH_EMAILS_LIMIT: u32 = 25; +/// 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; -/// Memory namespace prefix used when persisting sync snapshots. Mirrors -/// the `skill-{id}` convention in [`crate::openhuman::memory::store::client`] -/// so namespace listings stay coherent across composio + js skills. -const MEMORY_NAMESPACE: &str = "composio-gmail"; +/// 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; + +/// 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; @@ -59,8 +77,6 @@ impl ComposioProvider for GmailProvider { } fn sync_interval_secs(&self) -> Option { - // 15 minutes — matches the default `syncIntervalMinutes` the - // QuickJS gmail skill uses. Some(15 * 60) } @@ -88,11 +104,6 @@ impl ComposioProvider for GmailProvider { } let data = &resp.data; - // Composio wraps results in `{ data: { ... }, successful: bool }` - // and the upstream Gmail API returns `{ emailAddress, messagesTotal, - // threadsTotal, historyId }`. We dig through both `data` and the - // raw root because backend wrappers occasionally collapse the - // outer envelope. let email = pick_str( data, &[ @@ -124,10 +135,6 @@ impl ComposioProvider for GmailProvider { avatar_url: None, extras: data.clone(), }; - // PII discipline: never log the actual email address. We log - // only non-PII indicators (presence of an email, the domain - // portion if any) so the trace is still useful for debugging - // missing-profile cases without leaking the user's identity. let has_email = profile.email.is_some(); let email_domain = profile .email @@ -145,67 +152,233 @@ impl ComposioProvider for GmailProvider { async fn sync(&self, ctx: &ProviderContext, reason: SyncReason) -> Result { let started_at_ms = now_ms(); + let connection_id = ctx + .connection_id + .clone() + .unwrap_or_else(|| "default".to_string()); + tracing::info!( - connection_id = ?ctx.connection_id, + connection_id = %connection_id, reason = reason.as_str(), - "[composio:gmail] sync starting" + "[composio:gmail] incremental sync starting" ); - // For initial syncs, we ask for a slightly larger window so the - // first impression of the user's inbox is meaningful. Periodic - // ticks stay small. - let limit = match reason { - SyncReason::ConnectionCreated => FETCH_EMAILS_LIMIT * 2, - _ => FETCH_EMAILS_LIMIT, + // ── Step 1: load persistent sync state ────────────────────── + let Some(memory) = ctx.memory_client() else { + return Err("[composio:gmail] memory client not ready".to_string()); }; - let args = json!({ - "max_results": limit, - "query": "in:inbox -in:spam -in:trash", - }); + let mut state = SyncState::load(&memory, "gmail", &connection_id).await?; - let resp = ctx - .client - .execute_tool(ACTION_FETCH_EMAILS, Some(args)) - .await - .map_err(|e| format!("[composio:gmail] {ACTION_FETCH_EMAILS} failed: {e:#}"))?; - - 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}")); + // ── 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: now_ms(), + summary: "gmail sync skipped: daily budget exhausted".to_string(), + details: json!({ "budget_exhausted": true }), + }); } - let messages = extract_messages(&resp.data); - let items_ingested = persist_messages(ctx, &messages).await; - let finished_at_ms = now_ms(); + // ── Step 3: paginated incremental fetch ───────────────────── + let page_size = match reason { + SyncReason::ConnectionCreated => INITIAL_PAGE_SIZE, + _ => PAGE_SIZE, + }; + let mut total_fetched: usize = 0; + let mut total_persisted: usize = 0; + let mut newest_date: Option = None; + let mut page_token: Option = None; + + for page_num in 0..MAX_PAGES_PER_SYNC { + if state.budget_exhausted() { + tracing::info!( + page = page_num, + "[composio:gmail] budget exhausted mid-sync, stopping pagination" + ); + 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. + let mut query = "in:inbox -in:spam -in:trash".to_string(); + if let Some(ref cursor) = state.cursor { + if let Some(date_filter) = 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" + ); + } + } + + let mut args = json!({ + "max_results": page_size, + "query": query, + }); + if let Some(ref token) = page_token { + args["page_token"] = json!(token); + } + + let resp = ctx + .client + .execute_tool(ACTION_FETCH_EMAILS, Some(args)) + .await + .map_err(|e| { + format!("[composio:gmail] {ACTION_FETCH_EMAILS} page {page_num}: {e:#}") + })?; + + state.record_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}" + )); + } + + let messages = extract_messages(&resp.data); + total_fetched += messages.len(); + + if messages.is_empty() { + tracing::debug!( + page = page_num, + "[composio:gmail] empty page, stopping pagination" + ); + break; + } + + // ── Step 4: deduplicate and persist per-item ──────────── + let mut all_already_synced = true; + for msg in &messages { + let Some(msg_id) = extract_item_id(msg, MESSAGE_ID_PATHS) else { + tracing::debug!("[composio:gmail] message missing ID, skipping"); + continue; + }; + + // Track the newest date we've seen for cursor advancement. + if let Some(date_val) = extract_item_id(msg, MESSAGE_DATE_PATHS) { + if newest_date + .as_ref() + .map_or(true, |existing| date_val > *existing) + { + newest_date = Some(date_val); + } + } + + if state.is_synced(&msg_id) { + continue; + } + all_already_synced = false; + + // Build a human-readable title for this email document. + let subject = pick_str( + msg, + &[ + "subject", + "data.subject", + "payload.headers.Subject", + "snippet", + ], + ) + .unwrap_or_else(|| format!("Email {msg_id}")); + let doc_id = format!("composio-gmail-msg-{msg_id}"); + let title = format!("Gmail: {subject}"); + + match persist_single_item( + &memory, + "gmail", + &doc_id, + &title, + msg, + "gmail", + ctx.connection_id.as_deref(), + ) + .await + { + Ok(_) => { + state.mark_synced(&msg_id); + total_persisted += 1; + } + Err(e) => { + tracing::warn!( + msg_id = %msg_id, + error = %e, + "[composio:gmail] failed to persist message (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" + ); + break; + } + + // Check for next page token. + page_token = extract_page_token(&resp.data); + if page_token.is_none() { + tracing::debug!(page = page_num, "[composio:gmail] no next page token, done"); + break; + } + } + + // ── Step 5: advance cursor and save state ─────────────────── + if let Some(new_cursor) = newest_date { + state.advance_cursor(&new_cursor); + } + state.save(&memory).await?; + + let finished_at_ms = now_ms(); let summary = format!( - "gmail sync ({reason}): fetched {fetched} message(s), persisted {persisted}", + "gmail sync ({reason}): fetched {total_fetched}, persisted {total_persisted} new, \ + budget remaining {remaining}", reason = reason.as_str(), - fetched = messages.len(), - persisted = items_ingested, + remaining = state.budget_remaining(), ); tracing::info!( - connection_id = ?ctx.connection_id, + connection_id = %connection_id, elapsed_ms = finished_at_ms.saturating_sub(started_at_ms), - fetched = messages.len(), - persisted = items_ingested, - "[composio:gmail] sync complete" + total_fetched, + total_persisted, + budget_remaining = state.budget_remaining(), + "[composio:gmail] incremental sync complete" ); Ok(SyncOutcome { toolkit: "gmail".to_string(), - connection_id: ctx.connection_id.clone(), + connection_id: Some(connection_id), reason: reason.as_str().to_string(), - items_ingested, + items_ingested: total_persisted, started_at_ms, finished_at_ms, summary, details: json!({ - "messages_fetched": messages.len(), - "limit": limit, + "messages_fetched": total_fetched, + "messages_persisted": total_persisted, + "budget_remaining": state.budget_remaining(), + "cursor": state.cursor, + "synced_ids_total": state.synced_ids.len(), }), }) } @@ -222,14 +395,9 @@ impl ComposioProvider for GmailProvider { "[composio:gmail] on_trigger" ); - // Only react to message-arrival triggers — other gmail triggers - // (label changes, etc.) don't justify a full sync round-trip. if trigger.eq_ignore_ascii_case("GMAIL_NEW_GMAIL_MESSAGE") || trigger.eq_ignore_ascii_case("GMAIL_NEW_MESSAGE") { - // Best-effort incremental pull. Errors here are logged but - // not propagated — the trigger subscriber doesn't have a - // user-facing error surface to forward into. if let Err(e) = self.sync(ctx, SyncReason::Manual).await { tracing::warn!( error = %e, @@ -243,10 +411,7 @@ impl ComposioProvider for GmailProvider { // ── helpers ──────────────────────────────────────────────────────── -/// Walk the Composio response envelope and pull out a list of message -/// objects. Composio is inconsistent about whether the array lives at -/// `data.messages`, `messages`, or `data.data.messages`, so we try a -/// handful of common shapes before giving up. +/// Walk the Composio response envelope and pull out message objects. fn extract_messages(data: &Value) -> Vec { let candidates = [ data.pointer("/data/messages"), @@ -263,64 +428,43 @@ fn extract_messages(data: &Value) -> Vec { Vec::new() } -/// Persist a sync snapshot into the global memory store under the -/// `composio-gmail` namespace. Returns the number of items recorded -/// (currently always one document — the snapshot, not per-message -/// rows). Per-message ingestion can come later if/when we add an -/// agent surface that benefits from it. -async fn persist_messages(ctx: &ProviderContext, messages: &[Value]) -> usize { - let Some(client) = ctx.memory_client() else { - tracing::debug!("[composio:gmail] memory client not ready, skipping persist"); - return 0; - }; - if messages.is_empty() { - return 0; +/// Try to extract a pagination token from the API response. +fn extract_page_token(data: &Value) -> Option { + let candidates = [ + data.pointer("/data/nextPageToken"), + data.pointer("/nextPageToken"), + data.pointer("/data/data/nextPageToken"), + ]; + for cand in candidates.into_iter().flatten() { + if let Some(s) = cand.as_str() { + let trimmed = s.trim(); + if !trimmed.is_empty() { + return Some(trimmed.to_string()); + } + } } + None +} - let connection_label = ctx - .connection_id - .clone() - .unwrap_or_else(|| "default".to_string()); - let title = format!("gmail sync — {connection_label}"); - let snapshot = json!({ - "toolkit": "gmail", - "connection_id": ctx.connection_id, - "messages": messages, - "synced_at_ms": now_ms(), - }); - let content = serde_json::to_string_pretty(&snapshot).unwrap_or_else(|_| "{}".to_string()); - - if let Err(e) = client - .store_skill_sync( - // The store_skill_sync helper namespaces as `skill-{id}`, - // so we pass `gmail` here and rely on the standard prefix. - // The composio domain reads from `skill-gmail` namespaces - // through the same memory store as the JS gmail skill — - // intentional, so the agent's `recall_memory` sees both. - MEMORY_NAMESPACE.trim_start_matches("composio-"), - &connection_label, - &title, - &content, - Some("composio-sync".to_string()), - Some(json!({ - "toolkit": "gmail", - "connection_id": ctx.connection_id, - "source": "composio-provider", - })), - Some("medium".to_string()), - None, - None, - Some(format!("composio-gmail-{connection_label}")), - ) - .await - { - tracing::warn!( - error = %e, - "[composio:gmail] persist snapshot failed (non-fatal)" - ); - return 0; +/// 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. +fn cursor_to_gmail_after_filter(cursor: &str) -> Option { + // Try parsing as epoch millis first (Gmail's internalDate). + if let Ok(millis) = cursor.parse::() { + let secs = millis / 1000; + if let Some(dt) = chrono::DateTime::from_timestamp(secs, 0) { + return Some(dt.format("%Y/%m/%d").to_string()); + } } - 1 + // Try parsing as an ISO date/datetime. + if let Ok(dt) = chrono::NaiveDate::parse_from_str(cursor, "%Y-%m-%d") { + return Some(dt.format("%Y/%m/%d").to_string()); + } + if let Ok(dt) = chrono::DateTime::parse_from_rfc3339(cursor) { + return Some(dt.format("%Y/%m/%d").to_string()); + } + None } fn now_ms() -> u64 { @@ -356,6 +500,48 @@ mod tests { assert_eq!(extract_messages(&v).len(), 0); } + #[test] + fn extract_page_token_finds_nested() { + let v = json!({ "data": { "nextPageToken": "tok123" } }); + assert_eq!(extract_page_token(&v), Some("tok123".to_string())); + } + + #[test] + fn extract_page_token_none_when_missing() { + let v = json!({ "data": {} }); + assert_eq!(extract_page_token(&v), None); + } + + #[test] + fn cursor_to_filter_from_epoch_millis() { + // 2026-04-01 00:00:00 UTC in millis + let millis = "1774915200000"; + let filter = cursor_to_gmail_after_filter(millis); + assert!(filter.is_some()); + // Should produce a YYYY/MM/DD date. + let f = filter.unwrap(); + assert!(f.contains('/'), "Expected date with slashes, got {f}"); + } + + #[test] + fn cursor_to_filter_from_iso_date() { + assert_eq!( + cursor_to_gmail_after_filter("2026-03-15"), + Some("2026/03/15".to_string()) + ); + } + + #[test] + fn cursor_to_filter_from_rfc3339() { + let f = cursor_to_gmail_after_filter("2026-03-15T12:00:00Z"); + assert_eq!(f, Some("2026/03/15".to_string())); + } + + #[test] + fn cursor_to_filter_returns_none_for_garbage() { + assert_eq!(cursor_to_gmail_after_filter("not-a-date"), None); + } + #[test] fn provider_metadata_is_stable() { let p = GmailProvider::new(); diff --git a/src/openhuman/composio/providers/mod.rs b/src/openhuman/composio/providers/mod.rs index 092940c94..1a3fcf651 100644 --- a/src/openhuman/composio/providers/mod.rs +++ b/src/openhuman/composio/providers/mod.rs @@ -43,6 +43,7 @@ use super::client::{build_composio_client, ComposioClient}; pub mod gmail; pub mod notion; pub mod registry; +pub mod sync_state; pub use registry::{ all_providers, get_provider, init_default_providers, register_provider, ProviderArc, diff --git a/src/openhuman/composio/providers/notion.rs b/src/openhuman/composio/providers/notion.rs index 041496355..4d0fdd852 100644 --- a/src/openhuman/composio/providers/notion.rs +++ b/src/openhuman/composio/providers/notion.rs @@ -1,20 +1,24 @@ -//! Notion provider — native Rust counterpart to the QuickJS notion skill. +//! Notion provider — incremental sync with per-item persistence. //! -//! Behaves like [`super::gmail::GmailProvider`] but for Notion: pulls -//! the connected user's "about" record + a window of recent pages on -//! sync, persists snapshots into the global memory store, and reacts -//! to Notion triggers (typically `NOTION_PAGE_*` events) by re-running -//! the incremental sync. +//! On each sync pass: //! -//! Notion's Composio shape is intentionally squishy in this provider: -//! the upstream `users/me` and search endpoints have stable fields -//! (`name`, `person.email`, `results[]`), but Composio occasionally -//! re-wraps them. We use [`super::pick_str`] for tolerant extraction -//! so a minor backend change does not break the provider. +//! 1. Load persistent [`SyncState`] from the KV store. +//! 2. Check the daily request budget — bail early if exhausted. +//! 3. Fetch a page of recently edited pages via `NOTION_FETCH_DATA`, +//! sorted by `last_edited_time` descending. When a cursor exists +//! we can stop as soon as we see pages older than the cursor. +//! 4. Deduplicate against `synced_ids` in the state. Pages that have +//! been *edited* since their last sync are re-persisted (the cursor +//! is based on `last_edited_time`, so an edited page appears again). +//! 5. Persist each **new or updated** page as its own memory document. +//! 6. Paginate (up to budget) until no more results or all items in the +//! page are older than the cursor. +//! 7. Advance the cursor and save state. use async_trait::async_trait; use serde_json::{json, Value}; +use super::sync_state::{extract_item_id, persist_single_item, SyncState}; use super::{ pick_str, ComposioProvider, ProviderContext, ProviderUserProfile, SyncOutcome, SyncReason, }; @@ -22,7 +26,25 @@ use super::{ const ACTION_GET_ABOUT_ME: &str = "NOTION_GET_ABOUT_ME"; const ACTION_FETCH_DATA: &str = "NOTION_FETCH_DATA"; -const FETCH_LIMIT: u32 = 25; +/// Page size per API call. +const PAGE_SIZE: u32 = 25; + +/// Larger page size for initial sync after OAuth. +const INITIAL_PAGE_SIZE: u32 = 50; + +/// Maximum pages per sync pass. +const MAX_PAGES_PER_SYNC: u32 = 20; + +/// Paths for extracting a page's unique ID. +const PAGE_ID_PATHS: &[&str] = &["id", "data.id", "pageId", "data.pageId"]; + +/// Paths for extracting the `last_edited_time` used as sync cursor. +const PAGE_EDITED_PATHS: &[&str] = &[ + "last_edited_time", + "data.last_edited_time", + "lastEditedTime", + "data.lastEditedTime", +]; pub struct NotionProvider; @@ -45,8 +67,6 @@ impl ComposioProvider for NotionProvider { } fn sync_interval_secs(&self) -> Option { - // 30 minutes — Notion content changes less frequently than - // email, no need for the gmail cadence. Some(30 * 60) } @@ -114,76 +134,227 @@ impl ComposioProvider for NotionProvider { async fn sync(&self, ctx: &ProviderContext, reason: SyncReason) -> Result { let started_at_ms = now_ms(); + let connection_id = ctx + .connection_id + .clone() + .unwrap_or_else(|| "default".to_string()); + tracing::info!( - connection_id = ?ctx.connection_id, + connection_id = %connection_id, reason = reason.as_str(), - "[composio:notion] sync starting" + "[composio:notion] incremental sync starting" ); - let limit = match reason { - SyncReason::ConnectionCreated => FETCH_LIMIT * 2, - _ => FETCH_LIMIT, + // ── Step 1: load persistent sync state ────────────────────── + let Some(memory) = ctx.memory_client() else { + return Err("[composio:notion] memory client not ready".to_string()); }; - // NOTION_FETCH_DATA is a generic search/list action. We - // intentionally restrict to `object: page` and sort by - // `last_edited_time` descending so the sync pulls the most - // recently touched pages — that's what the agent's recall - // path benefits from most. Databases are skipped here on - // purpose: most users have far more pages than databases, - // and including databases would silently bloat the snapshot - // size for everyone. If we ever want to surface databases - // we should do it as a separate, opt-in fetch. - let args = json!({ - "page_size": limit, - "filter": { "value": "page", "property": "object" }, - "sort": { "direction": "descending", "timestamp": "last_edited_time" } - }); + let mut state = SyncState::load(&memory, "notion", &connection_id).await?; - let resp = ctx - .client - .execute_tool(ACTION_FETCH_DATA, Some(args)) - .await - .map_err(|e| format!("[composio:notion] {ACTION_FETCH_DATA} failed: {e:#}"))?; - - if !resp.successful { - let err = resp - .error - .clone() - .unwrap_or_else(|| "provider reported failure".to_string()); - return Err(format!("[composio:notion] {ACTION_FETCH_DATA}: {err}")); + // ── Step 2: check daily budget ────────────────────────────── + if state.budget_exhausted() { + tracing::info!( + connection_id = %connection_id, + "[composio:notion] daily request budget exhausted, skipping sync" + ); + return Ok(SyncOutcome { + toolkit: "notion".to_string(), + connection_id: Some(connection_id), + reason: reason.as_str().to_string(), + items_ingested: 0, + started_at_ms, + finished_at_ms: now_ms(), + summary: "notion sync skipped: daily budget exhausted".to_string(), + details: json!({ "budget_exhausted": true }), + }); } - let results = extract_results(&resp.data); - let items_ingested = persist_snapshot(ctx, &results) - .await - .map_err(|e| format!("[composio:notion] persist_snapshot failed: {e}"))?; - let finished_at_ms = now_ms(); + // ── Step 3: paginated incremental fetch ───────────────────── + let page_size = match reason { + SyncReason::ConnectionCreated => INITIAL_PAGE_SIZE, + _ => PAGE_SIZE, + }; + let mut total_fetched: usize = 0; + let mut total_persisted: usize = 0; + let mut newest_edited_time: Option = None; + let mut notion_cursor: Option = None; + + for page_num in 0..MAX_PAGES_PER_SYNC { + if state.budget_exhausted() { + tracing::info!( + page = page_num, + "[composio:notion] budget exhausted mid-sync, stopping pagination" + ); + break; + } + + let mut args = json!({ + "page_size": page_size, + "filter": { "value": "page", "property": "object" }, + "sort": { "direction": "descending", "timestamp": "last_edited_time" } + }); + if let Some(ref cursor) = notion_cursor { + args["start_cursor"] = json!(cursor); + } + + let resp = ctx + .client + .execute_tool(ACTION_FETCH_DATA, Some(args)) + .await + .map_err(|e| { + format!("[composio:notion] {ACTION_FETCH_DATA} page {page_num}: {e:#}") + })?; + + state.record_requests(1); + + if !resp.successful { + let err = resp + .error + .clone() + .unwrap_or_else(|| "provider reported failure".to_string()); + let _ = state.save(&memory).await; + return Err(format!( + "[composio:notion] {ACTION_FETCH_DATA} page {page_num}: {err}" + )); + } + + let results = extract_results(&resp.data); + total_fetched += results.len(); + + if results.is_empty() { + tracing::debug!( + page = page_num, + "[composio:notion] empty page, stopping pagination" + ); + break; + } + + // ── Step 4: deduplicate and persist per-item ──────────── + let mut hit_cursor_boundary = false; + for page in &results { + let Some(page_id) = extract_item_id(page, PAGE_ID_PATHS) else { + tracing::debug!("[composio:notion] page missing ID, skipping"); + continue; + }; + + let edited_time = extract_item_id(page, PAGE_EDITED_PATHS); + + // Track the newest edited time for cursor advancement. + if let Some(ref et) = edited_time { + if newest_edited_time + .as_ref() + .map_or(true, |existing| et > existing) + { + newest_edited_time = Some(et.clone()); + } + } + + // For Notion, a page can be *edited* after we last synced + // it. We use a composite key of page_id + edited_time to + // detect this: if the page_id is in synced_ids but the + // edited_time is newer than the cursor, we re-sync it. + let sync_key = match &edited_time { + Some(et) => format!("{page_id}@{et}"), + None => page_id.clone(), + }; + + // If the page's edited time is older than our cursor, + // we've caught up — everything beyond is already synced. + if let (Some(ref cursor), Some(ref et)) = (&state.cursor, &edited_time) { + if et <= cursor && state.is_synced(&sync_key) { + hit_cursor_boundary = true; + continue; + } + } + + if state.is_synced(&sync_key) { + continue; + } + + // Build a title from the page's properties. + let title_text = + extract_page_title(page).unwrap_or_else(|| format!("Notion page {page_id}")); + let doc_id = format!("composio-notion-page-{page_id}"); + let title = format!("Notion: {title_text}"); + + match persist_single_item( + &memory, + "notion", + &doc_id, + &title, + page, + "notion", + ctx.connection_id.as_deref(), + ) + .await + { + Ok(_) => { + state.mark_synced(&sync_key); + total_persisted += 1; + } + Err(e) => { + tracing::warn!( + page_id = %page_id, + error = %e, + "[composio:notion] failed to persist page (continuing)" + ); + } + } + } + + if hit_cursor_boundary { + tracing::debug!( + page = page_num, + "[composio:notion] reached cursor boundary, stopping" + ); + break; + } + + // Check for next page cursor from Notion API. + notion_cursor = extract_notion_cursor(&resp.data); + if notion_cursor.is_none() { + tracing::debug!(page = page_num, "[composio:notion] no next cursor, done"); + break; + } + } + + // ── Step 5: advance cursor and save state ─────────────────── + if let Some(new_cursor) = newest_edited_time { + state.advance_cursor(&new_cursor); + } + state.save(&memory).await?; + + let finished_at_ms = now_ms(); let summary = format!( - "notion sync ({reason}): fetched {fetched} item(s), persisted {persisted}", + "notion sync ({reason}): fetched {total_fetched}, persisted {total_persisted} new, \ + budget remaining {remaining}", reason = reason.as_str(), - fetched = results.len(), - persisted = items_ingested, + remaining = state.budget_remaining(), ); tracing::info!( - connection_id = ?ctx.connection_id, + connection_id = %connection_id, elapsed_ms = finished_at_ms.saturating_sub(started_at_ms), - fetched = results.len(), - persisted = items_ingested, - "[composio:notion] sync complete" + total_fetched, + total_persisted, + budget_remaining = state.budget_remaining(), + "[composio:notion] incremental sync complete" ); Ok(SyncOutcome { toolkit: "notion".to_string(), - connection_id: ctx.connection_id.clone(), + connection_id: Some(connection_id), reason: reason.as_str().to_string(), - items_ingested, + items_ingested: total_persisted, started_at_ms, finished_at_ms, summary, details: json!({ - "results_fetched": results.len(), - "page_size": limit, + "results_fetched": total_fetched, + "results_persisted": total_persisted, + "budget_remaining": state.budget_remaining(), + "cursor": state.cursor, + "synced_ids_total": state.synced_ids.len(), }), }) } @@ -199,10 +370,6 @@ impl ComposioProvider for NotionProvider { trigger = %trigger, "[composio:notion] on_trigger" ); - // Notion triggers all imply "something in the workspace - // changed", so any of them should kick a fresh incremental - // sync. Best-effort: we don't propagate errors out of the - // trigger path. if let Err(e) = self.sync(ctx, SyncReason::Manual).await { tracing::warn!( error = %e, @@ -215,6 +382,7 @@ impl ComposioProvider for NotionProvider { // ── helpers ──────────────────────────────────────────────────────── +/// Walk the Composio response envelope for Notion page results. fn extract_results(data: &Value) -> Vec { let candidates = [ data.pointer("/data/results"), @@ -231,48 +399,56 @@ fn extract_results(data: &Value) -> Vec { Vec::new() } -async fn persist_snapshot(ctx: &ProviderContext, results: &[Value]) -> Result { - let Some(client) = ctx.memory_client() else { - tracing::debug!("[composio:notion] memory client not ready, skipping persist"); - return Ok(0); - }; - if results.is_empty() { - return Ok(0); +/// Extract the Notion pagination cursor (for `start_cursor` on the +/// next request). +fn extract_notion_cursor(data: &Value) -> Option { + let candidates = [ + data.pointer("/data/next_cursor"), + data.pointer("/next_cursor"), + data.pointer("/data/data/next_cursor"), + ]; + for cand in candidates.into_iter().flatten() { + if let Some(s) = cand.as_str() { + let trimmed = s.trim(); + if !trimmed.is_empty() { + return Some(trimmed.to_string()); + } + } + } + None +} + +/// Try to extract a human-readable title from a Notion page object. +/// +/// Notion pages store the title in `properties.title` or +/// `properties.Name.title[0].plain_text`. We try several shapes. +fn extract_page_title(page: &Value) -> Option { + // Try the common `properties.title.title[0].plain_text` shape. + let props = page + .get("properties") + .or_else(|| page.get("data")?.get("properties")); + if let Some(props) = props { + // Walk all properties looking for a "title" type field. + if let Some(obj) = props.as_object() { + for (_key, val) in obj { + if val.get("type").and_then(Value::as_str) == Some("title") { + if let Some(arr) = val.get("title").and_then(Value::as_array) { + let text: String = arr + .iter() + .filter_map(|t| t.get("plain_text").and_then(Value::as_str)) + .collect::>() + .join(""); + if !text.is_empty() { + return Some(text); + } + } + } + } + } } - let connection_label = ctx - .connection_id - .clone() - .unwrap_or_else(|| "default".to_string()); - let title = format!("notion sync — {connection_label}"); - let snapshot = json!({ - "toolkit": "notion", - "connection_id": ctx.connection_id, - "results": results, - "synced_at_ms": now_ms(), - }); - let content = serde_json::to_string_pretty(&snapshot).unwrap_or_else(|_| "{}".to_string()); - - client - .store_skill_sync( - "notion", - &connection_label, - &title, - &content, - Some("composio-sync".to_string()), - Some(json!({ - "toolkit": "notion", - "connection_id": ctx.connection_id, - "source": "composio-provider", - })), - Some("medium".to_string()), - None, - None, - Some(format!("composio-notion-{connection_label}")), - ) - .await - .map_err(|e| format!("store_skill_sync: {e}"))?; - Ok(1) + // Fallback: top-level "title" field (some Composio shapes). + pick_str(page, &["title", "data.title", "name", "data.name"]) } fn now_ms() -> u64 { @@ -297,6 +473,50 @@ mod tests { assert_eq!(extract_results(&v3).len(), 0); } + #[test] + fn extract_notion_cursor_finds_nested() { + let v = json!({ "data": { "next_cursor": "abc123" } }); + assert_eq!(extract_notion_cursor(&v), Some("abc123".to_string())); + } + + #[test] + fn extract_notion_cursor_none_when_missing() { + let v = json!({ "data": { "has_more": false } }); + assert_eq!(extract_notion_cursor(&v), None); + } + + #[test] + fn extract_page_title_from_properties() { + let page = json!({ + "id": "page-1", + "properties": { + "Name": { + "type": "title", + "title": [ + { "plain_text": "My " }, + { "plain_text": "Page Title" } + ] + } + } + }); + assert_eq!(extract_page_title(&page), Some("My Page Title".to_string())); + } + + #[test] + fn extract_page_title_fallback_to_top_level() { + let page = json!({ "title": "Fallback Title" }); + assert_eq!( + extract_page_title(&page), + Some("Fallback Title".to_string()) + ); + } + + #[test] + fn extract_page_title_returns_none_when_missing() { + let page = json!({ "id": "p1" }); + assert_eq!(extract_page_title(&page), None); + } + #[test] fn provider_metadata_is_stable() { let p = NotionProvider::new(); diff --git a/src/openhuman/composio/providers/sync_state.rs b/src/openhuman/composio/providers/sync_state.rs new file mode 100644 index 000000000..e4768b17a --- /dev/null +++ b/src/openhuman/composio/providers/sync_state.rs @@ -0,0 +1,409 @@ +//! Persistent sync state for Composio providers. +//! +//! Each `(toolkit, connection_id)` pair gets its own [`SyncState`] persisted +//! in the local KV store. The state tracks: +//! +//! * **Cursor** — a provider-specific watermark (e.g. a timestamp or page +//! token) so the next sync can skip items already seen. +//! * **Synced IDs** — a set of item identifiers that have been written to +//! memory. Items in this set are skipped even if they appear again in +//! an API response (deduplication). +//! * **Daily request budget** — a rolling counter keyed by calendar date +//! (`YYYY-MM-DD`) that caps the number of `execute_tool` calls a +//! provider makes per day. Resets automatically when the date rolls +//! over. +//! +//! All persistence goes through [`crate::openhuman::memory::MemoryClient`]'s +//! KV surface (`kv_set` / `kv_get` under a dedicated namespace), so the +//! state survives process restarts without any extra file management. + +use std::collections::HashSet; + +use chrono::Utc; +use serde::{Deserialize, Serialize}; +use serde_json::json; + +use crate::openhuman::memory::MemoryClientRef; + +/// Maximum API requests a single provider connection may make per calendar +/// day. This covers the initial backfill case where there are thousands of +/// unsynced items — after this many requests the provider yields and +/// continues on the next day. +pub const DEFAULT_DAILY_REQUEST_LIMIT: u32 = 500; + +/// KV namespace under which all sync state keys live. Separate from the +/// memory document namespaces (`skill-gmail`, etc.) to avoid collisions. +const KV_NAMESPACE: &str = "composio-sync-state"; + +/// Persistent sync state for one `(toolkit, connection_id)` pair. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct SyncState { + /// Toolkit slug, e.g. `"gmail"`. + pub toolkit: String, + /// Connection id, e.g. `"conn_abc123"`. + pub connection_id: String, + + /// Provider-specific cursor. For Gmail this is the internal-date + /// (epoch millis) of the newest synced message; for Notion it is the + /// `last_edited_time` ISO string of the most recently synced page. + /// `None` means "never synced — start from scratch". + #[serde(default)] + pub cursor: Option, + + /// Set of item IDs that have already been persisted to memory. + /// Used for deduplication: if an item appears in an API response + /// but its ID is in this set, skip it. + #[serde(default)] + pub synced_ids: HashSet, + + /// Rolling daily request budget. + #[serde(default)] + pub daily_budget: DailyBudget, +} + +/// Tracks the number of API requests made on a given calendar day. +/// Automatically resets when the date rolls over. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct DailyBudget { + /// Calendar date in `YYYY-MM-DD` format. + pub date: String, + /// Number of `execute_tool` requests made so far today. + pub requests_used: u32, + /// Maximum requests allowed per day. + pub limit: u32, +} + +impl Default for DailyBudget { + fn default() -> Self { + Self { + date: today_str(), + requests_used: 0, + limit: DEFAULT_DAILY_REQUEST_LIMIT, + } + } +} + +impl DailyBudget { + /// Remaining requests available today. If the stored date is stale + /// (a previous day), this returns the full limit because the budget + /// will be reset on the next [`Self::record_request`] call. + pub fn remaining(&self) -> u32 { + if self.date != today_str() { + return self.limit; + } + self.limit.saturating_sub(self.requests_used) + } + + /// Returns `true` if the daily budget is exhausted for today. + pub fn is_exhausted(&self) -> bool { + self.remaining() == 0 + } + + /// Record `n` API requests. If the date has rolled over, resets the + /// counter before adding. + pub fn record_requests(&mut self, n: u32) { + let today = today_str(); + if self.date != today { + self.date = today; + self.requests_used = 0; + } + self.requests_used = self.requests_used.saturating_add(n); + } + + /// Record a single API request. + pub fn record_request(&mut self) { + self.record_requests(1); + } +} + +impl SyncState { + /// Create a fresh state for a new connection (never synced). + pub fn new(toolkit: impl Into, connection_id: impl Into) -> Self { + Self { + toolkit: toolkit.into(), + connection_id: connection_id.into(), + cursor: None, + synced_ids: HashSet::new(), + daily_budget: DailyBudget::default(), + } + } + + /// Whether the daily request budget is exhausted. + pub fn budget_exhausted(&self) -> bool { + self.daily_budget.is_exhausted() + } + + /// Remaining API requests for today. + pub fn budget_remaining(&self) -> u32 { + self.daily_budget.remaining() + } + + /// Record API requests made. + pub fn record_requests(&mut self, n: u32) { + self.daily_budget.record_requests(n); + } + + /// Check if an item ID has already been synced. + pub fn is_synced(&self, item_id: &str) -> bool { + self.synced_ids.contains(item_id) + } + + /// Mark an item ID as synced. + pub fn mark_synced(&mut self, item_id: impl Into) { + self.synced_ids.insert(item_id.into()); + } + + /// Update the cursor to a new watermark value. + pub fn advance_cursor(&mut self, cursor: impl Into) { + self.cursor = Some(cursor.into()); + } + + /// KV key for this state. Deterministic so load + save are symmetric. + fn kv_key(&self) -> String { + format!("{}:{}", self.toolkit, self.connection_id) + } + + /// Load sync state from the KV store, or return a fresh default if + /// none exists. + pub async fn load( + memory: &MemoryClientRef, + toolkit: &str, + connection_id: &str, + ) -> Result { + let key = format!("{toolkit}:{connection_id}"); + match memory.kv_get(Some(KV_NAMESPACE), &key).await? { + Some(value) => { + let mut state: SyncState = serde_json::from_value(value) + .map_err(|e| format!("[sync_state] deserialize failed for {key}: {e}"))?; + // Ensure budget rolls over if date changed. + if state.daily_budget.date != today_str() { + tracing::debug!( + toolkit, + connection_id, + old_date = %state.daily_budget.date, + "[sync_state] daily budget rolled over" + ); + state.daily_budget.date = today_str(); + state.daily_budget.requests_used = 0; + } + tracing::debug!( + toolkit, + connection_id, + cursor = ?state.cursor, + synced_ids_count = state.synced_ids.len(), + budget_remaining = state.budget_remaining(), + "[sync_state] loaded" + ); + Ok(state) + } + None => { + tracing::debug!( + toolkit, + connection_id, + "[sync_state] no existing state, starting fresh" + ); + Ok(Self::new(toolkit, connection_id)) + } + } + } + + /// Persist the current state to the KV store. + pub async fn save(&self, memory: &MemoryClientRef) -> Result<(), String> { + let key = self.kv_key(); + let value = serde_json::to_value(self) + .map_err(|e| format!("[sync_state] serialize failed: {e}"))?; + memory.kv_set(Some(KV_NAMESPACE), &key, &value).await?; + tracing::debug!( + toolkit = %self.toolkit, + connection_id = %self.connection_id, + cursor = ?self.cursor, + synced_ids_count = self.synced_ids.len(), + budget_used = self.daily_budget.requests_used, + "[sync_state] saved" + ); + Ok(()) + } +} + +/// Today's date as `YYYY-MM-DD` in UTC. +fn today_str() -> String { + Utc::now().format("%Y-%m-%d").to_string() +} + +/// Extract an ID string from a JSON value, trying multiple candidate paths. +/// Returns the first non-empty string found. +pub fn extract_item_id(item: &serde_json::Value, paths: &[&str]) -> Option { + for path in paths { + let mut cur = item; + let mut ok = true; + for segment in path.split('.') { + match cur.get(segment) { + Some(next) => cur = next, + None => { + ok = false; + break; + } + } + } + if !ok { + continue; + } + if let Some(s) = cur.as_str() { + let trimmed = s.trim(); + if !trimmed.is_empty() { + return Some(trimmed.to_string()); + } + } + } + None +} + +/// Helper to persist a single item as its own memory document. +/// +/// Each item is stored under the provider's memory namespace with a +/// deterministic `document_id` so repeated syncs upsert rather than +/// duplicate. Returns the document ID on success. +pub async fn persist_single_item( + memory: &MemoryClientRef, + namespace_skill_id: &str, + document_id: &str, + title: &str, + item: &serde_json::Value, + toolkit: &str, + connection_id: Option<&str>, +) -> Result { + let content = serde_json::to_string_pretty(item).unwrap_or_else(|_| "{}".to_string()); + memory + .store_skill_sync( + namespace_skill_id, + connection_id.unwrap_or("default"), + title, + &content, + Some("composio-sync".to_string()), + Some(json!({ + "toolkit": toolkit, + "connection_id": connection_id, + "source": "composio-provider-incremental", + })), + Some("medium".to_string()), + None, + None, + Some(document_id.to_string()), + ) + .await?; + Ok(document_id.to_string()) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn daily_budget_defaults_to_full() { + let b = DailyBudget::default(); + assert_eq!(b.remaining(), DEFAULT_DAILY_REQUEST_LIMIT); + assert!(!b.is_exhausted()); + } + + #[test] + fn daily_budget_tracks_requests() { + let mut b = DailyBudget::default(); + b.record_requests(100); + assert_eq!(b.remaining(), DEFAULT_DAILY_REQUEST_LIMIT - 100); + assert!(!b.is_exhausted()); + } + + #[test] + fn daily_budget_exhaustion() { + let mut b = DailyBudget::default(); + b.record_requests(DEFAULT_DAILY_REQUEST_LIMIT); + assert_eq!(b.remaining(), 0); + assert!(b.is_exhausted()); + } + + #[test] + fn daily_budget_saturates_on_overflow() { + let mut b = DailyBudget::default(); + b.record_requests(DEFAULT_DAILY_REQUEST_LIMIT + 100); + assert_eq!(b.remaining(), 0); + } + + #[test] + fn daily_budget_resets_on_date_change() { + let mut b = DailyBudget { + date: "2025-01-01".to_string(), + requests_used: 499, + limit: DEFAULT_DAILY_REQUEST_LIMIT, + }; + // Calling remaining() when date is stale returns full limit. + assert_eq!(b.remaining(), DEFAULT_DAILY_REQUEST_LIMIT); + // Recording a request resets the counter. + b.record_request(); + assert_eq!(b.date, today_str()); + assert_eq!(b.requests_used, 1); + } + + #[test] + fn sync_state_deduplication() { + let mut state = SyncState::new("gmail", "conn_1"); + assert!(!state.is_synced("msg_abc")); + state.mark_synced("msg_abc"); + assert!(state.is_synced("msg_abc")); + assert!(!state.is_synced("msg_xyz")); + } + + #[test] + fn sync_state_cursor_advancement() { + let mut state = SyncState::new("notion", "conn_2"); + assert!(state.cursor.is_none()); + state.advance_cursor("2026-04-01T00:00:00Z"); + assert_eq!(state.cursor.as_deref(), Some("2026-04-01T00:00:00Z")); + state.advance_cursor("2026-04-10T00:00:00Z"); + assert_eq!(state.cursor.as_deref(), Some("2026-04-10T00:00:00Z")); + } + + #[test] + fn sync_state_serialization_roundtrip() { + let mut state = SyncState::new("gmail", "conn_test"); + state.advance_cursor("12345"); + state.mark_synced("item_a"); + state.mark_synced("item_b"); + state.daily_budget.record_requests(42); + + let json = serde_json::to_value(&state).unwrap(); + let restored: SyncState = serde_json::from_value(json).unwrap(); + + assert_eq!(restored.toolkit, "gmail"); + assert_eq!(restored.connection_id, "conn_test"); + assert_eq!(restored.cursor.as_deref(), Some("12345")); + assert!(restored.synced_ids.contains("item_a")); + assert!(restored.synced_ids.contains("item_b")); + assert_eq!(restored.synced_ids.len(), 2); + assert_eq!(restored.daily_budget.requests_used, 42); + } + + #[test] + fn extract_item_id_walks_paths() { + let item = serde_json::json!({ + "id": "top_level", + "data": { "id": "nested" } + }); + assert_eq!( + extract_item_id(&item, &["data.id", "id"]), + Some("nested".to_string()) + ); + assert_eq!( + extract_item_id(&item, &["missing", "id"]), + Some("top_level".to_string()) + ); + assert_eq!(extract_item_id(&item, &["nope"]), None); + } + + #[test] + fn kv_key_is_deterministic() { + let s1 = SyncState::new("gmail", "conn_x"); + let s2 = SyncState::new("gmail", "conn_x"); + assert_eq!(s1.kv_key(), s2.kv_key()); + assert_eq!(s1.kv_key(), "gmail:conn_x"); + } +}