diff --git a/src/memory/sync/composio/mod.rs b/src/memory/sync/composio/mod.rs index 1af0a74..ff8563a 100644 --- a/src/memory/sync/composio/mod.rs +++ b/src/memory/sync/composio/mod.rs @@ -14,6 +14,6 @@ pub use connect::{ pub use gmail::GmailSyncPipeline; pub use orchestrator::{run_incremental_sync, IncrementalSource, PageFetch, SyncItem, SyncScope}; pub use providers::{ - ClickUpSyncPipeline, GitHubSyncPipeline, LinearSyncPipeline, NotionSyncPipeline, - SlackSearchBackfillPipeline, SlackSyncPipeline, + AsanaSyncPipeline, ClickUpSyncPipeline, GitHubSyncPipeline, LinearSyncPipeline, + NotionSyncPipeline, SlackSearchBackfillPipeline, SlackSyncPipeline, }; diff --git a/src/memory/sync/composio/providers/asana.rs b/src/memory/sync/composio/providers/asana.rs new file mode 100644 index 0000000..8ea7dee --- /dev/null +++ b/src/memory/sync/composio/providers/asana.rs @@ -0,0 +1,274 @@ +//! Incremental Asana synchronization through Composio. +//! +//! Asana is a task-shaped source: we enumerate the connected user's projects +//! (across every accessible workspace) and page each project's tasks. The +//! provider mirrors the GitHub pipeline's *server-side depth* strategy — Asana +//! filters by `modified_since`, so every fetched task is strictly newer than the +//! persisted cursor and the orchestrator's cursor-boundary check never truncates +//! an unordered page. Combined with a stable `asana:` document id, re-syncs +//! are idempotent (see `openhuman#4953`: the upsert key must be the stable id, +//! never a per-run id). + +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_WORKSPACES: &str = "ASANA_GET_MULTIPLE_WORKSPACES"; +const ACTION_PROJECTS: &str = "ASANA_GET_MULTIPLE_PROJECTS"; +const ACTION_TASKS: &str = "ASANA_GET_MULTIPLE_TASKS"; + +/// Asana returns only `gid`/`name`/`resource_type` unless `opt_fields` is set; +/// `modified_at` is required for the incremental cursor. +const TASK_OPT_FIELDS: &str = + "name,modified_at,created_at,completed,completed_at,notes,permalink_url,assignee.name,due_on"; + +pub struct AsanaSyncPipeline { + client: ComposioClient, + connection_id: String, + max_pages: usize, + page_size: usize, +} + +impl AsanaSyncPipeline { + pub fn new(client: ComposioClient, connection_id: impl Into) -> Self { + Self { + client, + connection_id: connection_id.into(), + max_pages: 20, + page_size: 50, + } + } + + /// Time floor for the incremental fetch: the persisted cursor once one + /// exists, otherwise the configured backfill depth. Formatted as the RFC + /// 3339 timestamp Asana's `modified_since` expects. + fn modified_since(&self, config: &MemoryConfig, state: &SyncState) -> Option { + if let Some(cursor) = state.cursor.as_deref() { + return Some(cursor.to_owned()); + } + config.sync.budget.sync_depth_days.map(|days| { + (chrono::Utc::now() - chrono::Duration::days(days as i64)) + .format("%Y-%m-%dT%H:%M:%SZ") + .to_string() + }) + } +} + +#[async_trait] +impl SyncPipeline for AsanaSyncPipeline { + fn id(&self) -> &str { + "composio:asana" + } + 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 AsanaSyncPipeline { + fn toolkit(&self) -> &'static str { + "asana" + } + fn action(&self) -> &'static str { + ACTION_TASKS + } + fn max_pages(&self) -> usize { + self.max_pages + } + /// Asana filters by `modified_since` server-side, so the engine must not + /// also apply a client-side depth floor. + fn server_side_depth(&self) -> bool { + true + } + /// A single unreachable project must not abort the whole sync. + fn tolerate_scope_errors(&self) -> bool { + true + } + async fn scopes( + &self, + executor: &dyn ActionExecutor, + connection_id: &str, + state: &mut SyncState, + ) -> anyhow::Result> { + let workspaces_response = checked_execute( + executor, + ACTION_WORKSPACES, + serde_json::json!({}), + connection_id, + state, + ) + .await?; + let workspaces = extract_records(&workspaces_response.data); + let mut scopes = Vec::new(); + for workspace in &workspaces { + let Some(workspace_id) = record_id(workspace) else { + continue; + }; + if state.budget_exhausted() { + break; + } + let projects_response = checked_execute( + executor, + ACTION_PROJECTS, + serde_json::json!({ + "workspace": workspace_id, + "archived": false, + "limit": 100, + "opt_fields": "name", + }), + connection_id, + state, + ) + .await?; + let projects = extract_records(&projects_response.data); + for project in &projects { + let Some(project_id) = record_id(project) else { + continue; + }; + let name = pick_str(project, &["name", "data.name"]) + .unwrap_or_else(|| format!("project {project_id}")); + scopes.push( + SyncScope::named(project_id.clone(), format!("project:{project_id}")) + .with_metadata(serde_json::json!({ + "workspace_id": workspace_id, + "project_name": name, + })), + ); + } + } + tracing::debug!( + toolkit = "asana", + connection_id, + workspaces = workspaces.len(), + projects = scopes.len(), + "[sync:asana] resolved project scopes" + ); + Ok(scopes) + } + fn arguments( + &self, + scope: &SyncScope, + config: &MemoryConfig, + state: &SyncState, + page: Option<&str>, + ) -> Value { + let mut args = serde_json::json!({ + "project": scope.id, + "limit": self.page_size, + "opt_fields": TASK_OPT_FIELDS, + }); + if let Some(since) = self.modified_since(config, state) { + args["modified_since"] = Value::String(since); + } + if let Some(offset) = page { + args["offset"] = Value::String(offset.to_owned()); + } + args + } + fn extract_page(&self, data: &Value, _: Option<&str>) -> PageFetch { + PageFetch { + items: extract_records(data), + next: [ + "/data/next_page/offset", + "/next_page/offset", + "/data/data/next_page/offset", + "/response_data/next_page/offset", + ] + .iter() + .find_map(|path| data.pointer(path).and_then(Value::as_str)) + .map(str::trim) + .filter(|offset| !offset.is_empty()) + .map(str::to_owned), + } + } + fn dedup_key(&self, item: &Value) -> Option { + let id = task_id(item)?; + Some(match self.sort_cursor(item) { + Some(modified) => format!("{id}@{modified}"), + None => id, + }) + } + fn sort_cursor(&self, item: &Value) -> Option { + pick_str( + item, + &[ + "modified_at", + "data.modified_at", + "modified_on", + "data.modified_on", + ], + ) + } + async fn document( + &self, + scope: &SyncScope, + connection_id: &str, + item: SyncItem, + _: &dyn ActionExecutor, + _: &mut SyncState, + ) -> anyhow::Result { + let id = task_id(&item.raw).unwrap_or_else(|| item.dedup_key.clone()); + let title = pick_str(&item.raw, &["name", "data.name", "title", "data.title"]) + .unwrap_or_else(|| format!("Asana task {id}")); + let content = serde_json::to_string_pretty(&item.raw)?; + // `document()` sets `document_id = asana:` and `taint = external_sync`. + let mut result = document("asana", connection_id, &id, title, content, item.raw); + // Stable collection scope for dedupe/path grouping — the project, never a + // per-run identifier. + result.metadata["path_scope"] = Value::String(format!("asana/project/{}", scope.id)); + result.metadata["project_id"] = Value::String(scope.id.clone()); + if let Some(workspace_id) = scope.metadata.get("workspace_id").and_then(Value::as_str) { + result.metadata["workspace_id"] = Value::String(workspace_id.to_owned()); + } + Ok(result) + } +} + +/// Asana wraps its collection payloads in `data`, and Composio wraps the +/// tool response again — so the task/project/workspace array may sit under +/// `/data/data`, `/data`, or the bare root depending on envelope nesting. +fn extract_records(data: &Value) -> Vec { + first_array( + data, + &[ + "/data/data", + "/data", + "/response_data/data", + "/data/response_data/data", + "/data/items", + "/items", + ], + ) +} + +fn record_id(record: &Value) -> Option { + pick_str(record, &["gid", "id", "data.gid", "data.id"]) +} + +fn task_id(item: &Value) -> Option { + pick_str(item, &["gid", "id", "data.gid", "data.id", "task_gid"]) +} + +#[cfg(test)] +#[path = "asana_tests.rs"] +mod tests; diff --git a/src/memory/sync/composio/providers/asana_tests.rs b/src/memory/sync/composio/providers/asana_tests.rs new file mode 100644 index 0000000..dedb41f --- /dev/null +++ b/src/memory/sync/composio/providers/asana_tests.rs @@ -0,0 +1,118 @@ +use serde_json::json; + +use super::*; +use crate::memory::config::{ComposioMode, ComposioSyncConfig, MemoryConfig, SecretString}; +use crate::memory::sync::state::SyncState; + +fn pipeline() -> AsanaSyncPipeline { + AsanaSyncPipeline::new( + ComposioClient::new(ComposioSyncConfig { + mode: ComposioMode::Direct, + base_url: "http://127.0.0.1:1".into(), + api_key: Some(SecretString::new("test-key")), + bearer_token: None, + entity_id: Some("entity-1".into()), + }), + "conn-asana", + ) +} + +#[test] +fn toolkit_and_action_use_asana_slugs() { + let pipeline = pipeline(); + assert_eq!(pipeline.toolkit(), "asana"); + assert_eq!(pipeline.action(), "ASANA_GET_MULTIPLE_TASKS"); + assert!(pipeline.server_side_depth()); +} + +#[test] +fn extract_page_reads_tasks_and_offset_cursor() { + let pipeline = pipeline(); + // Composio-wrapped Asana list payload: tasks under `data.data`, the paging + // token under `data.next_page.offset`. + let payload = json!({ + "data": { + "data": [ + {"gid": "1201", "name": "Ship sync", "modified_at": "2026-05-02T10:00:00.000Z"}, + {"gid": "1202", "name": "Write tests", "modified_at": "2026-05-01T09:00:00.000Z"} + ], + "next_page": {"offset": "eyJvIjoxfQ==", "path": "/tasks?offset=eyJvIjoxfQ=="} + } + }); + let page = pipeline.extract_page(&payload, None); + assert_eq!(page.items.len(), 2); + assert_eq!(page.next.as_deref(), Some("eyJvIjoxfQ==")); +} + +#[test] +fn extract_page_without_next_page_stops() { + let pipeline = pipeline(); + let payload = json!({"data": {"data": [{"gid": "9", "modified_at": "2026-01-01T00:00:00Z"}]}}); + let page = pipeline.extract_page(&payload, None); + assert_eq!(page.items.len(), 1); + assert_eq!(page.next, None); +} + +#[test] +fn dedup_key_pins_the_stable_gid_and_modified_version() { + let pipeline = pipeline(); + let task = + json!({"gid": "1201", "name": "Ship sync", "modified_at": "2026-05-02T10:00:00.000Z"}); + assert_eq!( + pipeline.dedup_key(&task).as_deref(), + Some("1201@2026-05-02T10:00:00.000Z") + ); + assert_eq!( + pipeline.sort_cursor(&task).as_deref(), + Some("2026-05-02T10:00:00.000Z") + ); +} + +#[tokio::test] +async fn document_uses_stable_gid_id_and_project_scope() { + let pipeline = pipeline(); + let scope = SyncScope::named("77", "project:77") + .with_metadata(json!({"workspace_id": "ws-9", "project_name": "Launch"})); + let raw = + json!({"gid": "1201", "name": "Ship sync", "modified_at": "2026-05-02T10:00:00.000Z"}); + let item = SyncItem { + dedup_key: "1201@2026-05-02T10:00:00.000Z".into(), + sort_cursor: Some("2026-05-02T10:00:00.000Z".into()), + raw, + }; + let client = ComposioClient::new(ComposioSyncConfig { + mode: ComposioMode::Direct, + base_url: "http://127.0.0.1:1".into(), + api_key: Some(SecretString::new("test-key")), + bearer_token: None, + entity_id: Some("entity-1".into()), + }); + let mut state = SyncState::new("asana", "conn-asana"); + let document = pipeline + .document(&scope, "conn-asana", item, &client, &mut state) + .await + .unwrap(); + + // Stable dedupe: id is the gid, not a per-run value; re-syncs upsert in place. + assert_eq!(document.document_id, "asana:1201"); + assert_eq!(document.title, "Ship sync"); + assert_eq!(document.toolkit, "asana"); + assert_eq!(document.metadata["taint"], "external_sync"); + assert_eq!(document.metadata["path_scope"], "asana/project/77"); + assert_eq!(document.metadata["project_id"], "77"); + assert_eq!(document.metadata["workspace_id"], "ws-9"); +} + +#[test] +fn arguments_carry_project_and_modified_since_from_cursor() { + let pipeline = pipeline(); + let scope = SyncScope::named("77", "project:77"); + let config = MemoryConfig::new("/tmp/tinycortex-asana-args"); + let mut state = SyncState::new("asana", "conn-asana"); + state.advance_cursor("2026-05-01T00:00:00Z"); + let args = pipeline.arguments(&scope, &config, &state, Some("offset-2")); + assert_eq!(args["project"], "77"); + assert_eq!(args["modified_since"], "2026-05-01T00:00:00Z"); + assert_eq!(args["offset"], "offset-2"); + assert_eq!(args["opt_fields"], TASK_OPT_FIELDS); +} diff --git a/src/memory/sync/composio/providers/mod.rs b/src/memory/sync/composio/providers/mod.rs index 2cbe5d1..13174d3 100644 --- a/src/memory/sync/composio/providers/mod.rs +++ b/src/memory/sync/composio/providers/mod.rs @@ -1,5 +1,6 @@ //! Incremental Composio provider pipelines. +mod asana; mod clickup; mod common; mod github; @@ -8,6 +9,7 @@ mod notion; mod slack; mod slack_parse; +pub use asana::AsanaSyncPipeline; pub use clickup::ClickUpSyncPipeline; pub use github::GitHubSyncPipeline; pub use linear::LinearSyncPipeline; diff --git a/src/memory/sync/mod.rs b/src/memory/sync/mod.rs index f9cf60f..3a0fd50 100644 --- a/src/memory/sync/mod.rs +++ b/src/memory/sync/mod.rs @@ -17,9 +17,10 @@ pub use audit::{ }; 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, - LinearSyncPipeline, NotionSyncPipeline, SlackSearchBackfillPipeline, SlackSyncPipeline, + resolve_auth_config_id, status_is_active, status_is_terminal, AsanaSyncPipeline, + ClickUpSyncPipeline, ComposioClient, ConnectionLink, EntityStore, GitHubSyncPipeline, + GmailSyncPipeline, 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 8c93d6f..95d44fe 100644 --- a/tests/composio_sync_mock.rs +++ b/tests/composio_sync_mock.rs @@ -6,10 +6,10 @@ use async_trait::async_trait; use serde_json::Value; use tinycortex::memory::config::{ComposioMode, ComposioSyncConfig, MemoryConfig, SecretString}; use tinycortex::memory::sync::{ - ClickUpSyncPipeline, ComposioClient, GitHubSyncPipeline, GmailSyncPipeline, LinearSyncPipeline, - NotionSyncPipeline, SkillDocSink, SkillDocument, SlackSearchBackfillPipeline, - SlackSyncPipeline, SyncContext, SyncEvent, SyncEventSink, SyncPipeline, SyncStage, SyncState, - SyncStateStore, + AsanaSyncPipeline, ClickUpSyncPipeline, ComposioClient, GitHubSyncPipeline, GmailSyncPipeline, + 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}; @@ -446,6 +446,65 @@ async fn clickup_pages_each_workspace_with_resolved_user() { ); } +#[tokio::test] +async fn asana_enumerates_projects_and_stores_stable_task_documents() { + let server = MockServer::start().await; + Mock::given(path("/tools/execute/ASANA_GET_MULTIPLE_WORKSPACES")) + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({ + "successful": true, + "data": {"data": [{"gid": "ws-9", "name": "Acme"}]} + }))) + .mount(&server) + .await; + Mock::given(path("/tools/execute/ASANA_GET_MULTIPLE_PROJECTS")) + .and(body_partial_json( + serde_json::json!({"arguments": {"workspace": "ws-9"}}), + )) + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({ + "successful": true, + "data": {"data": [{"gid": "proj-1", "name": "Launch"}]} + }))) + .mount(&server) + .await; + Mock::given(path("/tools/execute/ASANA_GET_MULTIPLE_TASKS")) + .and(body_partial_json( + serde_json::json!({"arguments": {"project": "proj-1"}}), + )) + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({ + "successful": true, + "data": {"data": [ + {"gid": "task-1", "name": "Ship sync", "modified_at": "2026-05-02T10:00:00.000Z"} + ]} + }))) + .mount(&server) + .await; + let (captures, context) = test_context(); + let pipeline = AsanaSyncPipeline::new( + ComposioClient::new(direct_config(server.uri(), "key")), + "asana-conn", + ); + let outcome = pipeline.tick(&test_config(), &context).await.unwrap(); + assert_eq!(outcome.records_ingested, 1); + assert_eq!(outcome.actions_called, 3); + { + let docs = captures.documents.lock().unwrap(); + assert_eq!(docs[0].document_id, "asana:task-1"); + assert_eq!(docs[0].metadata["taint"], "external_sync"); + assert_eq!(docs[0].metadata["path_scope"], "asana/project/proj-1"); + assert_eq!(docs[0].metadata["workspace_id"], "ws-9"); + } + let state = SyncState::load(captures.as_ref(), "asana", "asana-conn") + .await + .unwrap(); + assert!(state.is_synced("task-1@2026-05-02T10:00:00.000Z")); + assert_eq!(state.cursor.as_deref(), Some("2026-05-02T10:00:00.000Z")); + + // Re-sync must be idempotent: the same stable id is suppressed by dedup. + let second = pipeline.tick(&test_config(), &context).await.unwrap(); + assert_eq!(second.records_ingested, 0); + assert_eq!(captures.documents.lock().unwrap().len(), 1); +} + struct SlackHistory; impl Respond for SlackHistory {