admin/openhuman
1
0
Fork 0
mirror of https://github.com/tinyhumansai/openhuman.git synced 2026-07-27 21:08:00 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(composio): incremental sync with per-item persistence for Gmail and Notion (#519)

Browse Source 
* 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.
This commit is contained in:
Steven Enamakel
2026-04-12 17:58:51 -07:00
committed by GitHub
parent a115758c6c
commit 08178b5ff3
4 changed files with 1058 additions and 242 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/openhuman/composio/providers/gmail.rs
+323 -137
View File
@@ -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<u64> {
// 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<SyncOutcome, String> {
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<String> = None;
let mut page_token: Option<String> = 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<Value> {
let candidates = [
data.pointer("/data/messages"),
@@ -263,64 +428,43 @@ fn extract_messages(data: &Value) -> Vec<Value> {
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<String> {
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<String> {
// Try parsing as epoch millis first (Gmail's internalDate).
if let Ok(millis) = cursor.parse::<i64>() {
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();
src/openhuman/composio/providers/mod.rs
+1
View File
@@ -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,
src/openhuman/composio/providers/notion.rs
+325 -105
View File
@@ -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<u64> {
// 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<SyncOutcome, String> {
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<String> = None;
let mut notion_cursor: Option<String> = 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<Value> {
let candidates = [
data.pointer("/data/results"),
@@ -231,48 +399,56 @@ fn extract_results(data: &Value) -> Vec<Value> {
Vec::new()
}
async fn persist_snapshot(ctx: &ProviderContext, results: &[Value]) -> Result<usize, String> {
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<String> {
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<String> {
// 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::<Vec<_>>()
.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();
src/openhuman/composio/providers/sync_state.rs
+409
View File
@@ -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<String>,
/// 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<String>,
/// 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<String>, connection_id: impl Into<String>) -> 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<String>) {
self.synced_ids.insert(item_id.into());
}
/// Update the cursor to a new watermark value.
pub fn advance_cursor(&mut self, cursor: impl Into<String>) {
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<Self, String> {
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<String> {
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<String, String> {
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");
}
}