feat(sync): add Composio trello memory-sync pipeline

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,
 };

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;

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:<card id>"`,
+//! 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<String>) -> 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<SyncOutcome> {
+ 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<Vec<SyncScope>> {
+ 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<String> {
+ 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<String> {
+ 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<SkillDocument> {
+ 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;

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<ExecuteResponse> {
+ 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<serde_json::Value> = (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"));
+}

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;