diff --git a/src/memory/sync/composio/mod.rs b/src/memory/sync/composio/mod.rs index 4cf1e0b..9fa9422 100644 --- a/src/memory/sync/composio/mod.rs +++ b/src/memory/sync/composio/mod.rs @@ -14,6 +14,7 @@ pub use connect::{ pub use gmail::GmailSyncPipeline; pub use orchestrator::{run_incremental_sync, IncrementalSource, PageFetch, SyncItem, SyncScope}; pub use providers::{ - ClickUpSyncPipeline, GitHubSyncPipeline, GoogleCalendarSyncPipeline, GoogleDriveSyncPipeline, - LinearSyncPipeline, NotionSyncPipeline, SlackSearchBackfillPipeline, SlackSyncPipeline, + ClickUpSyncPipeline, GitHubSyncPipeline, GoogleCalendarSyncPipeline, GoogleDocsSyncPipeline, + GoogleDriveSyncPipeline, GoogleSheetsSyncPipeline, LinearSyncPipeline, NotionSyncPipeline, + SlackSearchBackfillPipeline, SlackSyncPipeline, }; diff --git a/src/memory/sync/composio/providers/google_docs.rs b/src/memory/sync/composio/providers/google_docs.rs new file mode 100644 index 0000000..269774b --- /dev/null +++ b/src/memory/sync/composio/providers/google_docs.rs @@ -0,0 +1,186 @@ +use async_trait::async_trait; +use serde_json::Value; + +use super::common::{checked_execute, document, first_array, pick_str}; +use crate::memory::config::MemoryConfig; +use crate::memory::sync::composio::{ + run_incremental_sync, ActionExecutor, ComposioClient, IncrementalSource, PageFetch, SyncItem, + SyncScope, +}; +use crate::memory::sync::state::SyncState; +use crate::memory::sync::traits::{ + SkillDocument, SyncContext, SyncOutcome, SyncPipeline, SyncPipelineKind, +}; + +const ACTION_SEARCH: &str = "GOOGLEDOCS_SEARCH_DOCUMENTS"; +const ACTION_PLAINTEXT: &str = "GOOGLEDOCS_GET_DOCUMENT_PLAINTEXT"; + +/// Incremental Google Docs synchronization through Composio. +/// +/// Two-step, document-shaped (like `NotionSyncPipeline`): `GOOGLEDOCS_SEARCH_DOCUMENTS` +/// enumerates accessible documents, then `GOOGLEDOCS_GET_DOCUMENT_PLAINTEXT` fetches the +/// body for each item inside [`IncrementalSource::document`]. +pub struct GoogleDocsSyncPipeline { + client: ComposioClient, + connection_id: String, + max_pages: usize, + page_size: usize, +} + +impl GoogleDocsSyncPipeline { + pub fn new(client: ComposioClient, connection_id: impl Into) -> Self { + Self { + client, + connection_id: connection_id.into(), + // NOTE: SEARCH_DOCUMENTS' page-token arg name is not pinned by the + // curated catalog, so we do a single-page-per-tick fetch (no page + // token emitted) rather than guessing a pagination scheme. Capped at + // 1 page: since `arguments()` never advances the token, a >1 cap + // would re-fire the identical page-1 request and burn budget slots + // for silently-deduplicated items. + max_pages: 1, + page_size: 25, + } + } +} + +#[async_trait] +impl SyncPipeline for GoogleDocsSyncPipeline { + fn id(&self) -> &str { + "composio:googledocs" + } + fn kind(&self) -> SyncPipelineKind { + SyncPipelineKind::Composio + } + async fn init(&self, _: &MemoryConfig, _: &SyncContext) -> anyhow::Result<()> { + Ok(()) + } + async fn tick( + &self, + config: &MemoryConfig, + context: &SyncContext, + ) -> anyhow::Result { + run_incremental_sync(self, &self.client, &self.connection_id, config, context).await + } +} + +#[async_trait] +impl IncrementalSource for GoogleDocsSyncPipeline { + fn toolkit(&self) -> &'static str { + "googledocs" + } + fn action(&self) -> &'static str { + ACTION_SEARCH + } + fn max_pages(&self) -> usize { + self.max_pages + } + fn arguments( + &self, + _: &SyncScope, + _: &MemoryConfig, + _: &SyncState, + _page: Option<&str>, + ) -> Value { + // NOTE: an empty/broad `query` enumerates every accessible document; + // `max_results` bounds the batch. Both mirror the underlying Drive + // search parameters. No page token is emitted (see `max_pages`). + serde_json::json!({"query": "", "max_results": self.page_size}) + } + fn extract_page(&self, data: &Value, _: Option<&str>) -> PageFetch { + PageFetch { + items: first_array( + data, + &[ + "/data/documents", + "/documents", + "/data/files", + "/files", + "/data/results", + "/results", + "/data/items", + "/items", + ], + ), + // Bounded fetch: no page token consumed (see `max_pages`). The + // pointers are read defensively should Composio surface one. + next: [ + "/data/nextPageToken", + "/nextPageToken", + "/data/next_page_token", + "/next_page_token", + ] + .iter() + .find_map(|path| data.pointer(path).and_then(Value::as_str)) + .map(str::trim) + .filter(|token| !token.is_empty()) + .map(str::to_owned), + } + } + fn dedup_key(&self, item: &Value) -> Option { + let id = pick_str(item, &["id", "data.id", "documentId", "data.documentId"])?; + Some(match self.sort_cursor(item) { + Some(modified) => format!("{id}@{modified}"), + None => id, + }) + } + fn sort_cursor(&self, item: &Value) -> Option { + pick_str( + item, + &[ + "modifiedTime", + "data.modifiedTime", + "modified_time", + "updatedTime", + ], + ) + } + async fn document( + &self, + _: &SyncScope, + connection_id: &str, + item: SyncItem, + executor: &dyn ActionExecutor, + state: &mut SyncState, + ) -> anyhow::Result { + let id = pick_str( + &item.raw, + &["id", "data.id", "documentId", "data.documentId"], + ) + .unwrap_or_else(|| item.dedup_key.clone()); + let title = pick_str(&item.raw, &["title", "data.title", "name", "data.name"]) + .unwrap_or_else(|| format!("Google Doc {id}")); + // NOTE: GET_DOCUMENT_PLAINTEXT identifies the doc by an id argument; + // Composio commonly keys this as "id" (or "document_id"). We send "id". + let response = checked_execute( + executor, + ACTION_PLAINTEXT, + serde_json::json!({"id": id}), + connection_id, + state, + ) + .await?; + let content = [ + "/data/text", + "/text", + "/data/plaintext", + "/plaintext", + "/data/content", + "/content", + "/data/response_data/text", + ] + .iter() + .find_map(|path| response.data.pointer(path).and_then(Value::as_str)) + .filter(|value| !value.trim().is_empty()) + .map(str::to_owned) + .unwrap_or(serde_json::to_string_pretty(&item.raw)?); + Ok(document( + "googledocs", + connection_id, + &id, + title, + content, + item.raw, + )) + } +} diff --git a/src/memory/sync/composio/providers/google_sheets.rs b/src/memory/sync/composio/providers/google_sheets.rs new file mode 100644 index 0000000..bc8317a --- /dev/null +++ b/src/memory/sync/composio/providers/google_sheets.rs @@ -0,0 +1,186 @@ +use async_trait::async_trait; +use serde_json::Value; + +use super::common::{checked_execute, document, first_array, pick_str}; +use crate::memory::config::MemoryConfig; +use crate::memory::sync::composio::{ + run_incremental_sync, ActionExecutor, ComposioClient, IncrementalSource, PageFetch, SyncItem, + SyncScope, +}; +use crate::memory::sync::state::SyncState; +use crate::memory::sync::traits::{ + SkillDocument, SyncContext, SyncOutcome, SyncPipeline, SyncPipelineKind, +}; + +const ACTION_SEARCH: &str = "GOOGLESHEETS_SEARCH_SPREADSHEETS"; +const ACTION_INFO: &str = "GOOGLESHEETS_GET_SPREADSHEET_INFO"; + +/// Incremental Google Sheets synchronization through Composio. +/// +/// Two-step, document-shaped (like `NotionSyncPipeline`): `GOOGLESHEETS_SEARCH_SPREADSHEETS` +/// enumerates accessible spreadsheets, then `GOOGLESHEETS_GET_SPREADSHEET_INFO` fetches the +/// spreadsheet metadata for each item inside [`IncrementalSource::document`]. +pub struct GoogleSheetsSyncPipeline { + client: ComposioClient, + connection_id: String, + max_pages: usize, + page_size: usize, +} + +impl GoogleSheetsSyncPipeline { + pub fn new(client: ComposioClient, connection_id: impl Into) -> Self { + Self { + client, + connection_id: connection_id.into(), + // NOTE: SEARCH_SPREADSHEETS' page-token arg name is not pinned by the + // curated catalog, so we do a single-page-per-tick fetch (no page + // token emitted) rather than guessing a pagination scheme. Capped at + // 1 page: since `arguments()` never advances the token, a >1 cap + // would re-fire the identical page-1 request and burn budget slots + // for silently-deduplicated items. + max_pages: 1, + page_size: 25, + } + } +} + +#[async_trait] +impl SyncPipeline for GoogleSheetsSyncPipeline { + fn id(&self) -> &str { + "composio:googlesheets" + } + fn kind(&self) -> SyncPipelineKind { + SyncPipelineKind::Composio + } + async fn init(&self, _: &MemoryConfig, _: &SyncContext) -> anyhow::Result<()> { + Ok(()) + } + async fn tick( + &self, + config: &MemoryConfig, + context: &SyncContext, + ) -> anyhow::Result { + run_incremental_sync(self, &self.client, &self.connection_id, config, context).await + } +} + +#[async_trait] +impl IncrementalSource for GoogleSheetsSyncPipeline { + fn toolkit(&self) -> &'static str { + "googlesheets" + } + fn action(&self) -> &'static str { + ACTION_SEARCH + } + fn max_pages(&self) -> usize { + self.max_pages + } + fn arguments( + &self, + _: &SyncScope, + _: &MemoryConfig, + _: &SyncState, + _page: Option<&str>, + ) -> Value { + // NOTE: an empty/broad `query` enumerates every accessible spreadsheet; + // `max_results` bounds the batch. Both mirror the underlying Drive + // search parameters. No page token is emitted (see `max_pages`). + serde_json::json!({"query": "", "max_results": self.page_size}) + } + fn extract_page(&self, data: &Value, _: Option<&str>) -> PageFetch { + PageFetch { + items: first_array( + data, + &[ + "/data/spreadsheets", + "/spreadsheets", + "/data/files", + "/files", + "/data/results", + "/results", + "/data/items", + "/items", + ], + ), + // Bounded fetch: no page token consumed (see `max_pages`). The + // pointers are read defensively should Composio surface one. + next: [ + "/data/nextPageToken", + "/nextPageToken", + "/data/next_page_token", + "/next_page_token", + ] + .iter() + .find_map(|path| data.pointer(path).and_then(Value::as_str)) + .map(str::trim) + .filter(|token| !token.is_empty()) + .map(str::to_owned), + } + } + fn dedup_key(&self, item: &Value) -> Option { + let id = pick_str( + item, + &["id", "data.id", "spreadsheetId", "data.spreadsheetId"], + )?; + Some(match self.sort_cursor(item) { + Some(modified) => format!("{id}@{modified}"), + None => id, + }) + } + fn sort_cursor(&self, item: &Value) -> Option { + pick_str( + item, + &["modifiedTime", "data.modifiedTime", "modified_time"], + ) + } + async fn document( + &self, + _: &SyncScope, + connection_id: &str, + item: SyncItem, + executor: &dyn ActionExecutor, + state: &mut SyncState, + ) -> anyhow::Result { + let id = pick_str( + &item.raw, + &["id", "data.id", "spreadsheetId", "data.spreadsheetId"], + ) + .unwrap_or_else(|| item.dedup_key.clone()); + let title = pick_str( + &item.raw, + &[ + "title", + "data.title", + "properties.title", + "data.properties.title", + "name", + ], + ) + .unwrap_or_else(|| format!("Google Sheet {id}")); + // NOTE: GET_SPREADSHEET_INFO identifies the spreadsheet by a + // "spreadsheet_id" argument (Google's canonical parameter name). + let response = checked_execute( + executor, + ACTION_INFO, + serde_json::json!({"spreadsheet_id": id}), + connection_id, + state, + ) + .await?; + // `response.data` is the already-unwrapped payload; the pointers catch + // any additional Composio wrapping, else we serialize the payload root. + let info = ["/data", "/data/data"] + .iter() + .find_map(|path| response.data.pointer(path)) + .unwrap_or(&response.data); + let content = serde_json::to_string_pretty(info)?; + Ok(document( + "googlesheets", + connection_id, + &id, + title, + content, + item.raw, + )) + } +} diff --git a/src/memory/sync/composio/providers/mod.rs b/src/memory/sync/composio/providers/mod.rs index 6847e2f..8d5bf84 100644 --- a/src/memory/sync/composio/providers/mod.rs +++ b/src/memory/sync/composio/providers/mod.rs @@ -4,7 +4,9 @@ mod clickup; mod common; mod github; mod google_calendar; +mod google_docs; mod google_drive; +mod google_sheets; mod linear; mod notion; mod slack; @@ -13,7 +15,9 @@ mod slack_parse; pub use clickup::ClickUpSyncPipeline; pub use github::GitHubSyncPipeline; pub use google_calendar::GoogleCalendarSyncPipeline; +pub use google_docs::GoogleDocsSyncPipeline; pub use google_drive::GoogleDriveSyncPipeline; +pub use google_sheets::GoogleSheetsSyncPipeline; pub use linear::LinearSyncPipeline; pub use notion::NotionSyncPipeline; pub use slack::{SlackSearchBackfillPipeline, SlackSyncPipeline}; diff --git a/src/memory/sync/mod.rs b/src/memory/sync/mod.rs index da31d9d..9fe63e3 100644 --- a/src/memory/sync/mod.rs +++ b/src/memory/sync/mod.rs @@ -19,8 +19,9 @@ pub use composio::{ create_connection_link, generate_entity_id, get_connection_status, list_auth_configs, resolve_auth_config_id, status_is_active, status_is_terminal, ClickUpSyncPipeline, ComposioClient, ConnectionLink, EntityStore, GitHubSyncPipeline, GmailSyncPipeline, - GoogleCalendarSyncPipeline, GoogleDriveSyncPipeline, LinearSyncPipeline, NotionSyncPipeline, - SlackSearchBackfillPipeline, SlackSyncPipeline, + GoogleCalendarSyncPipeline, GoogleDocsSyncPipeline, GoogleDriveSyncPipeline, + GoogleSheetsSyncPipeline, LinearSyncPipeline, NotionSyncPipeline, SlackSearchBackfillPipeline, + SlackSyncPipeline, }; pub use dispatcher::{SyncDispatcher, SyncRunResult}; pub use github::GithubRepoSyncPipeline; diff --git a/tests/composio_sync_mock.rs b/tests/composio_sync_mock.rs index 119e441..a0e6158 100644 --- a/tests/composio_sync_mock.rs +++ b/tests/composio_sync_mock.rs @@ -7,9 +7,10 @@ use serde_json::Value; use tinycortex::memory::config::{ComposioMode, ComposioSyncConfig, MemoryConfig, SecretString}; use tinycortex::memory::sync::{ ClickUpSyncPipeline, ComposioClient, GitHubSyncPipeline, GmailSyncPipeline, - GoogleCalendarSyncPipeline, GoogleDriveSyncPipeline, LinearSyncPipeline, NotionSyncPipeline, - SkillDocSink, SkillDocument, SlackSearchBackfillPipeline, SlackSyncPipeline, SyncContext, - SyncEvent, SyncEventSink, SyncPipeline, SyncStage, SyncState, SyncStateStore, + GoogleCalendarSyncPipeline, GoogleDocsSyncPipeline, GoogleDriveSyncPipeline, + GoogleSheetsSyncPipeline, LinearSyncPipeline, NotionSyncPipeline, SkillDocSink, SkillDocument, + SlackSearchBackfillPipeline, SlackSyncPipeline, SyncContext, SyncEvent, SyncEventSink, + SyncPipeline, SyncStage, SyncState, SyncStateStore, }; use wiremock::matchers::{body_partial_json, header, method, path, path_regex}; use wiremock::{Mock, MockServer, Request, Respond, ResponseTemplate}; @@ -412,6 +413,83 @@ async fn notion_fetches_markdown_and_counts_both_requests() { assert_eq!(state.daily_budget.requests_used, 2); } +#[tokio::test] +async fn google_docs_fetches_plaintext_and_counts_both_requests() { + let server = MockServer::start().await; + Mock::given(path("/tools/execute/GOOGLEDOCS_SEARCH_DOCUMENTS")) + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({ + "successful": true, + "data": {"documents": [{"id": "doc-1", "title": "Design", "modifiedTime": "2026-03-01T00:00:00Z"}]} + }))) + .mount(&server) + .await; + Mock::given(path("/tools/execute/GOOGLEDOCS_GET_DOCUMENT_PLAINTEXT")) + .and(body_partial_json( + serde_json::json!({"arguments": {"id": "doc-1"}}), + )) + .respond_with(ResponseTemplate::new(200).set_body_json( + serde_json::json!({"successful": true, "data": {"text": "Full plaintext body"}}), + )) + .mount(&server) + .await; + let (captures, context) = test_context(); + let pipeline = GoogleDocsSyncPipeline::new( + ComposioClient::new(direct_config(server.uri(), "key")), + "gdocs-conn", + ); + pipeline.tick(&test_config(), &context).await.unwrap(); + { + let documents = captures.documents.lock().unwrap(); + assert_eq!(documents[0].content, "Full plaintext body"); + assert_eq!(documents[0].document_id, "googledocs:doc-1"); + assert_eq!(documents[0].title, "Design"); + assert_eq!(documents[0].metadata["taint"], "external_sync"); + } + let state = SyncState::load(captures.as_ref(), "googledocs", "gdocs-conn") + .await + .unwrap(); + assert_eq!(state.daily_budget.requests_used, 2); +} + +#[tokio::test] +async fn google_sheets_fetches_info_and_stores_document() { + let server = MockServer::start().await; + Mock::given(path("/tools/execute/GOOGLESHEETS_SEARCH_SPREADSHEETS")) + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({ + "successful": true, + "data": {"spreadsheets": [{"id": "sheet-1", "title": "Budget", "modifiedTime": "2026-04-01T00:00:00Z"}]} + }))) + .mount(&server) + .await; + Mock::given(path("/tools/execute/GOOGLESHEETS_GET_SPREADSHEET_INFO")) + .and(body_partial_json( + serde_json::json!({"arguments": {"spreadsheet_id": "sheet-1"}}), + )) + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({ + "successful": true, + "data": {"properties": {"title": "Budget"}, "sheets": [{"properties": {"title": "Q1"}}]} + }))) + .mount(&server) + .await; + let (captures, context) = test_context(); + let pipeline = GoogleSheetsSyncPipeline::new( + ComposioClient::new(direct_config(server.uri(), "key")), + "gsheets-conn", + ); + pipeline.tick(&test_config(), &context).await.unwrap(); + { + let documents = captures.documents.lock().unwrap(); + assert_eq!(documents[0].document_id, "googlesheets:sheet-1"); + assert_eq!(documents[0].title, "Budget"); + assert_eq!(documents[0].metadata["taint"], "external_sync"); + assert!(documents[0].content.contains("Q1")); + } + let state = SyncState::load(captures.as_ref(), "googlesheets", "gsheets-conn") + .await + .unwrap(); + assert_eq!(state.daily_budget.requests_used, 2); +} + #[tokio::test] async fn clickup_pages_each_workspace_with_resolved_user() { let server = MockServer::start().await;