diff --git a/src/memory/sync/composio/mod.rs b/src/memory/sync/composio/mod.rs index 1af0a74..7fa4419 100644 --- a/src/memory/sync/composio/mod.rs +++ b/src/memory/sync/composio/mod.rs @@ -15,5 +15,5 @@ pub use gmail::GmailSyncPipeline; pub use orchestrator::{run_incremental_sync, IncrementalSource, PageFetch, SyncItem, SyncScope}; pub use providers::{ ClickUpSyncPipeline, GitHubSyncPipeline, LinearSyncPipeline, NotionSyncPipeline, - SlackSearchBackfillPipeline, SlackSyncPipeline, + SlackSearchBackfillPipeline, SlackSyncPipeline, TrelloSyncPipeline, }; diff --git a/src/memory/sync/composio/providers/mod.rs b/src/memory/sync/composio/providers/mod.rs index 2cbe5d1..2c33134 100644 --- a/src/memory/sync/composio/providers/mod.rs +++ b/src/memory/sync/composio/providers/mod.rs @@ -7,9 +7,11 @@ mod linear; mod notion; mod slack; mod slack_parse; +mod trello; pub use clickup::ClickUpSyncPipeline; pub use github::GitHubSyncPipeline; pub use linear::LinearSyncPipeline; pub use notion::NotionSyncPipeline; pub use slack::{SlackSearchBackfillPipeline, SlackSyncPipeline}; +pub use trello::TrelloSyncPipeline; diff --git a/src/memory/sync/composio/providers/trello.rs b/src/memory/sync/composio/providers/trello.rs new file mode 100644 index 0000000..3ba40ee --- /dev/null +++ b/src/memory/sync/composio/providers/trello.rs @@ -0,0 +1,191 @@ +//! Incremental Trello synchronization through Composio. +//! +//! Trello is a card-shaped source: each board is a [`SyncScope`], and cards on a +//! board are the syncable items. The pipeline models the ClickUp provider +//! (scope-per-container, global recency cursor) rather than Gmail, because +//! ingestion fans out over boards and pages cards within each one. +//! +//! Dedupe is keyed on the stable card id (`document_id = "trello:"`, +//! set by [`super::common::document`]); per-run identifiers are never used as the +//! upsert key. The dedup key additionally suffixes `dateLastActivity` so an +//! edited card re-ingests, mirroring the other incremental providers. + +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, +}; + +/// Boards visible to the authorized member (`idMember = "me"`). +const ACTION_BOARDS: &str = "TRELLO_GET_MEMBERS_BOARDS_BY_ID_MEMBER"; +/// Cards on a single board — the per-scope fetch action. +const ACTION_CARDS: &str = "TRELLO_GET_BOARDS_CARDS_BY_ID_BOARD"; + +pub struct TrelloSyncPipeline { + client: ComposioClient, + connection_id: String, + max_pages: usize, + page_size: usize, +} + +impl TrelloSyncPipeline { + pub fn new(client: ComposioClient, connection_id: impl Into) -> Self { + Self { + client, + connection_id: connection_id.into(), + max_pages: 20, + page_size: 50, + } + } +} + +#[async_trait] +impl SyncPipeline for TrelloSyncPipeline { + fn id(&self) -> &str { + "composio:trello" + } + 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 TrelloSyncPipeline { + fn toolkit(&self) -> &'static str { + "trello" + } + fn action(&self) -> &'static str { + ACTION_CARDS + } + fn max_pages(&self) -> usize { + self.max_pages + } + /// Boards are independent scopes: an inaccessible or archived board must not + /// abort ingestion for the rest. + fn tolerate_scope_errors(&self) -> bool { + true + } + async fn scopes( + &self, + executor: &dyn ActionExecutor, + connection_id: &str, + state: &mut SyncState, + ) -> anyhow::Result> { + let response = checked_execute( + executor, + ACTION_BOARDS, + serde_json::json!({"idMember": "me"}), + connection_id, + state, + ) + .await?; + let boards = first_array( + &response.data, + &["/data", "/data/items", "/items", "/data/boards", "/boards"], + ); + Ok(boards + .into_iter() + .filter_map(|board| pick_str(&board, &["id", "data.id"])) + .map(|id| SyncScope::named(id.clone(), format!("board:{id}"))) + .collect()) + } + fn arguments( + &self, + scope: &SyncScope, + _: &MemoryConfig, + _: &SyncState, + page: Option<&str>, + ) -> Value { + // Trello paginates card listings with `before`, a card id that bounds the + // window to cards older than it (ids are time-ordered). + let mut args = serde_json::json!({"idBoard": scope.id, "limit": self.page_size}); + if let Some(page) = page { + args["before"] = serde_json::json!(page); + } + args + } + fn extract_page(&self, data: &Value, _: Option<&str>) -> PageFetch { + // Composio wraps the Trello REST array under `data`; tolerate the common + // envelope shapes seen across toolkits. + let items = first_array( + data, + &[ + "/data", + "/data/items", + "/items", + "/data/data", + "/data/cards", + "/cards", + "/data/results", + "/results", + ], + ); + // Trello has no page token: request the next window with `before` set to + // the oldest card id in this full page. A short page means the board is + // drained. + let next = (items.len() == self.page_size) + .then(|| { + items + .last() + .and_then(|card| pick_str(card, &["id", "data.id"])) + }) + .flatten(); + PageFetch { items, next } + } + fn dedup_key(&self, item: &Value) -> Option { + let id = pick_str(item, &["id", "data.id"])?; + Some(match self.sort_cursor(item) { + Some(activity) => format!("{id}@{activity}"), + None => id, + }) + } + fn sort_cursor(&self, item: &Value) -> Option { + pick_str( + item, + &[ + "dateLastActivity", + "data.dateLastActivity", + "date_last_activity", + "data.date_last_activity", + ], + ) + } + async fn document( + &self, + scope: &SyncScope, + connection_id: &str, + item: SyncItem, + _: &dyn ActionExecutor, + _: &mut SyncState, + ) -> anyhow::Result { + let id = pick_str(&item.raw, &["id", "data.id"]).unwrap_or_else(|| item.dedup_key.clone()); + let title = pick_str(&item.raw, &["name", "data.name", "title", "data.title"]) + .unwrap_or_else(|| format!("Trello card {id}")); + let content = serde_json::to_string_pretty(&item.raw)?; + let mut result = document("trello", connection_id, &id, title, content, item.raw); + result.metadata["board_id"] = Value::String(scope.id.clone()); + Ok(result) + } +} + +#[cfg(test)] +#[path = "trello_tests.rs"] +mod tests; diff --git a/src/memory/sync/composio/providers/trello_tests.rs b/src/memory/sync/composio/providers/trello_tests.rs new file mode 100644 index 0000000..13838cc --- /dev/null +++ b/src/memory/sync/composio/providers/trello_tests.rs @@ -0,0 +1,104 @@ +use async_trait::async_trait; +use serde_json::json; + +use super::*; +use crate::memory::config::{ComposioMode, ComposioSyncConfig, SecretString}; +use crate::memory::sync::composio::client::{ActionExecutor, ExecuteResponse}; +use crate::memory::sync::composio::ComposioClient; + +/// Executor that must never be reached: `document()` for Trello serializes the +/// card in-place and issues no follow-up action call. +struct UnusedExecutor; + +#[async_trait] +impl ActionExecutor for UnusedExecutor { + async fn execute( + &self, + action: &str, + _arguments: serde_json::Value, + _connection_id: Option<&str>, + ) -> anyhow::Result { + panic!("document() unexpectedly executed action {action}"); + } +} + +fn pipeline() -> TrelloSyncPipeline { + let config = ComposioSyncConfig { + mode: ComposioMode::Direct, + base_url: "https://example.invalid".into(), + api_key: Some(SecretString::new("test-key")), + bearer_token: None, + entity_id: None, + }; + TrelloSyncPipeline::new(ComposioClient::new(config), "trello-conn") +} + +#[test] +fn toolkit_and_action_are_stable_slugs() { + let pipeline = pipeline(); + assert_eq!(pipeline.toolkit(), "trello"); + assert_eq!(pipeline.action(), "TRELLO_GET_BOARDS_CARDS_BY_ID_BOARD"); +} + +#[test] +fn extract_page_reads_cards_and_pages_on_full_window() { + let pipeline = pipeline(); + // A full page (page_size cards) yields a `before` cursor = oldest card id. + let cards: Vec = (0..pipeline.page_size) + .map(|i| json!({"id": format!("card-{i}"), "name": format!("Card {i}")})) + .collect(); + let last_id = format!("card-{}", pipeline.page_size - 1); + let payload = json!({"data": cards}); + + let page = pipeline.extract_page(&payload, None); + assert_eq!(page.items.len(), pipeline.page_size); + assert_eq!(page.next.as_deref(), Some(last_id.as_str())); + assert_eq!(page.items[0]["id"], json!("card-0")); + + // A short page drains the board: no further pagination. + let short = json!({"data": [{"id": "card-x", "name": "solo"}]}); + assert!(pipeline.extract_page(&short, None).next.is_none()); +} + +#[test] +fn dedup_key_suffixes_activity_for_reingest() { + let pipeline = pipeline(); + let card = json!({"id": "abc123", "dateLastActivity": "2026-07-20T10:00:00Z"}); + assert_eq!( + pipeline.dedup_key(&card).as_deref(), + Some("abc123@2026-07-20T10:00:00Z") + ); + let no_activity = json!({"id": "abc123"}); + assert_eq!(pipeline.dedup_key(&no_activity).as_deref(), Some("abc123")); +} + +#[tokio::test] +async fn document_uses_stable_card_id_as_document_id() { + let pipeline = pipeline(); + let scope = SyncScope::named("board-9", "board:board-9"); + let raw = json!({ + "id": "card-42", + "name": "Ship the pipeline", + "dateLastActivity": "2026-07-20T10:00:00Z" + }); + let mut state = SyncState::new("trello", "trello-conn"); + let item = SyncItem { + dedup_key: "card-42@2026-07-20T10:00:00Z".into(), + sort_cursor: Some("2026-07-20T10:00:00Z".into()), + raw: raw.clone(), + }; + + let doc = pipeline + .document(&scope, "trello-conn", item, &UnusedExecutor, &mut state) + .await + .expect("document conversion"); + + // Stable dedupe: the upsert key is derived from the card id, never a per-run id. + assert_eq!(doc.document_id, "trello:card-42"); + assert_eq!(doc.namespace_skill_id, "trello"); + assert_eq!(doc.toolkit, "trello"); + assert_eq!(doc.title, "Ship the pipeline"); + assert_eq!(doc.metadata["taint"], json!("external_sync")); + assert_eq!(doc.metadata["provider_id"], json!("card-42")); + assert_eq!(doc.metadata["board_id"], json!("board-9")); +} diff --git a/src/memory/sync/mod.rs b/src/memory/sync/mod.rs index f9cf60f..704808f 100644 --- a/src/memory/sync/mod.rs +++ b/src/memory/sync/mod.rs @@ -20,6 +20,7 @@ pub use composio::{ resolve_auth_config_id, status_is_active, status_is_terminal, ClickUpSyncPipeline, ComposioClient, ConnectionLink, EntityStore, GitHubSyncPipeline, GmailSyncPipeline, LinearSyncPipeline, NotionSyncPipeline, SlackSearchBackfillPipeline, SlackSyncPipeline, + TrelloSyncPipeline, }; pub use dispatcher::{SyncDispatcher, SyncRunResult}; pub use github::GithubRepoSyncPipeline;