From 2791638818f9599bf28b1fc6cd8d2ce2325ad31f Mon Sep 17 00:00:00 2001 From: sanil-23 Date: Mon, 22 Jun 2026 05:57:56 +0530 Subject: [PATCH] refactor(memory_sync): convert ClickUp to the generic sync orchestrator (Refs #3333, stacked on #3860) (#3873) Co-authored-by: Claude Co-authored-by: Steven Enamakel --- .../composio/providers/clickup/mod.rs | 1 + .../composio/providers/clickup/provider.rs | 738 +----------------- .../composio/providers/clickup/source.rs | 569 ++++++++++++++ .../memory_sync_providers_raw_coverage_e2e.rs | 81 ++ 4 files changed, 662 insertions(+), 727 deletions(-) create mode 100644 src/openhuman/memory_sync/composio/providers/clickup/source.rs diff --git a/src/openhuman/memory_sync/composio/providers/clickup/mod.rs b/src/openhuman/memory_sync/composio/providers/clickup/mod.rs index 5bb012498..949298230 100644 --- a/src/openhuman/memory_sync/composio/providers/clickup/mod.rs +++ b/src/openhuman/memory_sync/composio/providers/clickup/mod.rs @@ -15,6 +15,7 @@ mod ingest; mod provider; +mod source; mod sync; #[cfg(test)] mod tests; diff --git a/src/openhuman/memory_sync/composio/providers/clickup/provider.rs b/src/openhuman/memory_sync/composio/providers/clickup/provider.rs index 6688290e2..3dfb9d0db 100644 --- a/src/openhuman/memory_sync/composio/providers/clickup/provider.rs +++ b/src/openhuman/memory_sync/composio/providers/clickup/provider.rs @@ -25,12 +25,10 @@ //! and avoids accidentally ingesting other teammates' private tasks. use async_trait::async_trait; -use futures::StreamExt; -use serde_json::{json, Value}; +use serde_json::json; -use super::{ingest::ingest_task_into_memory_tree, sync}; -use crate::openhuman::config::Config; -use crate::openhuman::memory_sync::composio::providers::sync_state::{extract_item_id, SyncState}; +use super::source::run_clickup_sync; +use super::sync; use crate::openhuman::memory_sync::composio::providers::{ first_array_str, merge_extra, pick_str, resolve_sync_interval_secs, ComposioProvider, CuratedTool, NormalizedTask, ProviderContext, ProviderUserProfile, SyncOutcome, SyncReason, @@ -42,27 +40,9 @@ pub(crate) const ACTION_GET_AUTHORIZED_TEAMS_WORKSPACES: &str = "CLICKUP_GET_AUTHORIZED_TEAMS_WORKSPACES"; pub(crate) const ACTION_GET_FILTERED_TEAM_TASKS: &str = "CLICKUP_GET_FILTERED_TEAM_TASKS"; -/// Page size per API call. ClickUp's filtered-team-tasks endpoint -/// returns up to 100 tasks per page; we ask for a smaller window on -/// steady-state syncs to keep response sizes bounded. -const PAGE_SIZE: u32 = 50; - -/// Larger initial-sync page size, used immediately after OAuth so the -/// first backfill catches up faster. -const INITIAL_PAGE_SIZE: u32 = 100; - -/// Maximum pages (per workspace) per sync pass before yielding. Caps -/// initial backfill churn — anything beyond this rolls over to the -/// next sync interval. -const MAX_PAGES_PER_WORKSPACE: u32 = 20; - /// Paths for extracting a task's unique ID. Composio sometimes wraps /// the upstream payload under `data`, so we check both shapes. -const TASK_ID_PATHS: &[&str] = &["id", "data.id", "task_id", "data.task_id"]; - -/// Max in-flight ingests per page. DB writes serialize anyway and the -/// cloud embedder has rate limits, so keep this small. -const INGEST_CONCURRENCY: usize = 8; +pub(super) const TASK_ID_PATHS: &[&str] = &["id", "data.id", "task_id", "data.task_id"]; pub struct ClickUpProvider; @@ -150,340 +130,14 @@ impl ComposioProvider for ClickUpProvider { }) } + /// Incremental sync via the generic + /// [`orchestrator`](crate::openhuman::memory_sync::composio::providers::orchestrator): + /// user/workspace resolution, the per-workspace page loop, dedup, the + /// `max_items` cap, the epoch-ms `sync_depth_days` window, and cursor + /// handling live in `run_sync`; the ClickUp-specific primitives live in + /// [`super::source`]. async fn sync(&self, ctx: &ProviderContext, reason: SyncReason) -> Result { - let started_at_ms = sync::now_ms(); - let connection_id = ctx - .connection_id - .clone() - .unwrap_or_else(|| "default".to_string()); - - tracing::info!( - connection_id = %connection_id, - reason = reason.as_str(), - "[composio:clickup] incremental sync starting" - ); - - // ── Step 1: load persistent sync state ────────────────────── - let Some(memory) = ctx.memory_client() else { - return Err("[composio:clickup] memory client not ready".to_string()); - }; - let mut state = SyncState::load(&memory, "clickup", &connection_id).await?; - - // ── Step 2: check daily budget ────────────────────────────── - if state.budget_exhausted() { - tracing::info!( - connection_id = %connection_id, - "[composio:clickup] daily request budget exhausted, skipping sync" - ); - return Ok(SyncOutcome { - toolkit: "clickup".to_string(), - connection_id: Some(connection_id), - reason: reason.as_str().to_string(), - items_ingested: 0, - started_at_ms, - finished_at_ms: sync::now_ms(), - summary: "clickup sync skipped: daily budget exhausted".to_string(), - details: json!({ "budget_exhausted": true }), - }); - } - - // ── Step 3: resolve the authenticated user's numeric ID ───── - // - // ClickUp's "filtered team tasks" endpoint accepts an - // `assignees` filter as a list of user IDs. We need the - // *current* user's ID to scope the sync to "my tasks" rather - // than "everyone's tasks". The ID is stable for the lifetime - // of the OAuth connection, so we only fetch it once per sync - // pass (and cheaply re-fetch each pass — Composio caches and - // the call is sub-100ms). - let user_id = match self.resolve_user_id(ctx, &mut state).await { - Ok(id) => id, - Err(e) => { - let _ = state.save(&memory).await; - return Err(e); - } - }; - - // Re-check the budget here — `resolve_user_id` just spent one - // request, and if that pushed us over the cap, firing - // `CLICKUP_GET_AUTHORIZED_TEAMS_WORKSPACES` would be wasted - // work. Bailing here keeps the per-day API call count strictly - // honoured even when we entered the sync with one slot left. - if state.budget_exhausted() { - tracing::info!( - connection_id = %connection_id, - "[composio:clickup] budget exhausted after user-id probe, skipping sync" - ); - state.save(&memory).await?; - return Ok(SyncOutcome { - toolkit: "clickup".to_string(), - connection_id: Some(connection_id), - reason: reason.as_str().to_string(), - items_ingested: 0, - started_at_ms, - finished_at_ms: sync::now_ms(), - summary: "clickup sync skipped: daily budget exhausted after user-id probe" - .to_string(), - details: json!({ "budget_exhausted": true, "user_id_resolved": true }), - }); - } - - // ── Step 4: resolve which workspaces (teams) to iterate ───── - let workspaces = match self.resolve_workspaces(ctx, &mut state).await { - Ok(ws) => ws, - Err(e) => { - let _ = state.save(&memory).await; - return Err(e); - } - }; - - if workspaces.is_empty() { - tracing::info!( - connection_id = %connection_id, - "[composio:clickup] no workspaces visible to this connection; nothing to sync" - ); - state.save(&memory).await?; - return Ok(SyncOutcome { - toolkit: "clickup".to_string(), - connection_id: Some(connection_id), - reason: reason.as_str().to_string(), - items_ingested: 0, - started_at_ms, - finished_at_ms: sync::now_ms(), - summary: "clickup sync: no workspaces visible".to_string(), - details: json!({ "workspaces_visible": 0 }), - }); - } - - // ── Step 5: paginated incremental fetch per workspace ─────── - let page_size = match reason { - SyncReason::ConnectionCreated => INITIAL_PAGE_SIZE, - _ => PAGE_SIZE, - }; - - // ctx.max_items: route through ItemCap — page ceiling, mid-page - // per-item break, and post-page hard stop all share one source of truth. - let mut cap = super::super::helpers::ItemCap::new(ctx.max_items); - let effective_max_pages = cap.max_pages(page_size, MAX_PAGES_PER_WORKSPACE); - if ctx.max_items.is_some() && effective_max_pages < MAX_PAGES_PER_WORKSPACE { - tracing::debug!( - connection_id = %connection_id, - max_items = ?ctx.max_items, - effective_max_pages, - "[composio:clickup] [memory_sync] applying max_items page cap" - ); - } - - // ctx.sync_depth_days: oldest allowed date_updated for client-side skip. - // ClickUp's `date_updated` field is a millisecond-epoch string, so the - // floor must also be epoch-millis (not RFC3339) for the lexicographic - // compare in `select_pending` to work correctly. - let oldest_allowed_time: Option = ctx.sync_depth_days.map(|days| { - let floor = chrono::Utc::now() - chrono::Duration::days(days as i64); - let s = floor.timestamp_millis().to_string(); - tracing::debug!( - connection_id = %connection_id, - sync_depth_days = days, - oldest_allowed_ms = %s, - "[composio:clickup] [memory_sync] applying sync_depth_days floor" - ); - s - }); - - let mut total_fetched: usize = 0; - let mut total_persisted: usize = 0; - let mut newest_updated: Option = None; - let mut hit_cap_boundary = false; - - 'workspaces: for workspace_id in &workspaces { - for page_num in 0..effective_max_pages { - if state.budget_exhausted() { - tracing::info!( - workspace_id = %workspace_id, - page = page_num, - "[composio:clickup] budget exhausted mid-sync, stopping pagination" - ); - break 'workspaces; - } - - let args = json!({ - "team_id": workspace_id, - "assignees": [user_id.clone()], - "order_by": "updated", - "reverse": true, - "page": page_num, - "page_size": page_size, - // Include subtasks so per-list "checklist" style work - // also reaches Memory Tree. - "subtasks": true, - // Include archived = false (default) — we don't want - // closed-and-archived noise in memory. - }); - - let resp = ctx - .execute(ACTION_GET_FILTERED_TEAM_TASKS, Some(args)) - .await - .map_err(|e| { - format!( - "[composio:clickup] {ACTION_GET_FILTERED_TEAM_TASKS} \ - workspace={workspace_id} 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:clickup] {ACTION_GET_FILTERED_TEAM_TASKS} \ - workspace={workspace_id} page={page_num}: {err}" - )); - } - - let tasks = sync::extract_tasks(&resp.data); - total_fetched += tasks.len(); - - if tasks.is_empty() { - tracing::debug!( - workspace_id = %workspace_id, - page = page_num, - "[composio:clickup] empty page, moving to next workspace" - ); - break; - } - - // ── Per-item dedup + bounded-concurrency ingest ───── - let (mut pending, mut hit_cursor_boundary) = - select_pending(&tasks, &state, &mut newest_updated); - - // ctx.sync_depth_days: drop items updated before the depth floor. `pending` is - // in descending timestamp order, so truncate at the first item below the floor - // and signal cursor-boundary so pagination stops. - if let Some(ref floor) = oldest_allowed_time { - if let Some(cut) = pending.iter().position(|p| { - p.updated - .as_deref() - .map(|t| t < floor.as_str()) - .unwrap_or(false) - }) { - pending.truncate(cut); - hit_cursor_boundary = true; - } - } - - // ctx.max_items: clamp the dedup'd batch to the remaining budget before ingest. - cap.clamp_batch(&mut pending); - - let ingestor = MemoryTreeIngestor { - config: ctx.config.as_ref(), - connection_id: &connection_id, - }; - let outcome = - ingest_pending_buffered(&ingestor, pending, workspace_id, INGEST_CONCURRENCY) - .await; - for key in &outcome.synced_keys { - state.mark_synced(key); - } - total_persisted += outcome.persisted; - cap.record(outcome.persisted); - - // ctx.max_items precise cap: once the per-source cap is hit, stop paginating. - if cap.is_reached() { - hit_cap_boundary = true; - break 'workspaces; - } - - if hit_cursor_boundary { - tracing::debug!( - workspace_id = %workspace_id, - page = page_num, - "[composio:clickup] reached cursor boundary, stopping workspace" - ); - break; - } - - // ctx.max_items hard stop. - if cap.is_reached() { - tracing::debug!( - workspace_id = %workspace_id, - page = page_num, - total_persisted, - "[composio:clickup] [memory_sync] max_items reached, stopping pagination" - ); - hit_cap_boundary = true; - break 'workspaces; - } - - // ClickUp's filtered-team-tasks endpoint signals the last - // page implicitly: when fewer than `page_size` results - // come back, there are no more pages. - if (tasks.len() as u32) < page_size { - tracing::debug!( - workspace_id = %workspace_id, - page = page_num, - returned = tasks.len(), - "[composio:clickup] short page, end of workspace" - ); - break; - } - } - } - - // ── Step 6: advance cursor and save state ─────────────────── - // Hold the cursor on a cap-truncated pass so the next sync re-scans the unseen tail. - if !hit_cap_boundary { - if let Some(new_cursor) = newest_updated { - state.advance_cursor(&new_cursor); - } - } else { - tracing::warn!( - connection_id = %connection_id, - "[composio:clickup] holding cursor — cap-truncated pass; next sync will re-scan \ - the unseen tail" - ); - } - state.set_last_sync_at_ms(sync::now_ms()); - state.save(&memory).await?; - - let finished_at_ms = sync::now_ms(); - let summary = format!( - "clickup sync ({reason}): fetched {total_fetched}, persisted {total_persisted} new, \ - across {workspace_count} workspace(s), budget remaining {remaining}", - reason = reason.as_str(), - workspace_count = workspaces.len(), - remaining = state.budget_remaining(), - ); - tracing::info!( - connection_id = %connection_id, - elapsed_ms = finished_at_ms.saturating_sub(started_at_ms), - total_fetched, - total_persisted, - workspace_count = workspaces.len(), - budget_remaining = state.budget_remaining(), - "[composio:clickup] incremental sync complete" - ); - - Ok(SyncOutcome { - toolkit: "clickup".to_string(), - connection_id: Some(connection_id), - reason: reason.as_str().to_string(), - items_ingested: total_persisted, - started_at_ms, - finished_at_ms, - summary, - details: json!({ - "tasks_fetched": total_fetched, - "tasks_persisted": total_persisted, - "workspaces_visible": workspaces.len(), - "budget_remaining": state.budget_remaining(), - "cursor": state.cursor, - "synced_ids_total": state.synced_ids.len(), - }), - }) + run_clickup_sync(ctx, reason).await } async fn fetch_tasks( @@ -625,373 +279,3 @@ fn normalize_clickup_task(task: &serde_json::Value) -> Option { raw: task.clone(), }) } - -impl ClickUpProvider { - /// Look up (and budget-record) the authorized user's numeric ID. - /// - /// The ID is stable for the connection's lifetime, but we re-fetch - /// on every sync rather than persisting it because (a) the ClickUp - /// API call is cheap, (b) caching it in `SyncState` would inflate - /// the public struct for a single provider's quirk, and (c) it - /// implicitly validates that the OAuth connection is still good - /// before we start paginating. - async fn resolve_user_id( - &self, - ctx: &ProviderContext, - state: &mut SyncState, - ) -> Result { - let resp = ctx - .execute(ACTION_GET_AUTHORIZED_USER, Some(json!({}))) - .await - .map_err(|e| { - format!("[composio:clickup] {ACTION_GET_AUTHORIZED_USER} failed: {e:#}") - })?; - state.record_requests(1); - - if !resp.successful { - let err = resp - .error - .clone() - .unwrap_or_else(|| "provider reported failure".to_string()); - return Err(format!( - "[composio:clickup] {ACTION_GET_AUTHORIZED_USER}: {err}" - )); - } - - sync::extract_user_id(&resp.data).ok_or_else(|| { - "[composio:clickup] CLICKUP_GET_AUTHORIZED_USER returned no user.id".to_string() - }) - } - - /// Look up the list of workspace (team) IDs visible to this - /// connection. ClickUp's per-team task-filter endpoint requires a - /// concrete `team_id`, so we have to enumerate. - async fn resolve_workspaces( - &self, - ctx: &ProviderContext, - state: &mut SyncState, - ) -> Result, String> { - let resp = ctx - .execute(ACTION_GET_AUTHORIZED_TEAMS_WORKSPACES, Some(json!({}))) - .await - .map_err(|e| { - format!("[composio:clickup] {ACTION_GET_AUTHORIZED_TEAMS_WORKSPACES} failed: {e:#}") - })?; - state.record_requests(1); - - if !resp.successful { - let err = resp - .error - .clone() - .unwrap_or_else(|| "provider reported failure".to_string()); - return Err(format!( - "[composio:clickup] {ACTION_GET_AUTHORIZED_TEAMS_WORKSPACES}: {err}" - )); - } - - Ok(sync::extract_workspace_ids(&resp.data)) - } -} - -/// One task that passed dedupe in pass 1 and is queued for concurrent -/// ingest in pass 2. Borrows the raw task `Value` out of the current -/// page's `tasks` (same scope — no clone needed). -struct PendingIngest<'a> { - sync_key: String, - task_id: String, - title: String, - updated: Option, - task: &'a Value, -} - -/// Folded result of [`ingest_pending_buffered`]. Both fields are -/// order-independent, so the concurrent stage can accumulate into it -/// regardless of the order ingests complete. -#[derive(Default)] -struct BufferedIngestOutcome { - /// `sync_key`s whose ingest succeeded — the caller marks each synced. - synced_keys: Vec, - /// Number of tasks persisted (equals `synced_keys.len()`). - persisted: usize, -} - -/// Seam over "ingest one ClickUp task", so the bounded-concurrency driver -/// can be unit-tested with a fake that records peak in-flight calls -/// without a real memory tree or embedder. -#[async_trait] -trait TaskIngestor { - async fn ingest( - &self, - task_id: &str, - title: &str, - updated: Option<&str>, - task: &Value, - ) -> anyhow::Result; -} - -/// Production ingestor: routes into the memory-tree pipeline via -/// [`ingest_task_into_memory_tree`]. -struct MemoryTreeIngestor<'c> { - config: &'c Config, - connection_id: &'c str, -} - -#[async_trait] -impl TaskIngestor for MemoryTreeIngestor<'_> { - async fn ingest( - &self, - task_id: &str, - title: &str, - updated: Option<&str>, - task: &Value, - ) -> anyhow::Result { - ingest_task_into_memory_tree( - self.config, - self.connection_id, - task_id, - title, - updated, - task, - ) - .await - } -} - -/// Pass 1 (pure, no I/O): scan one page of `tasks`, advance -/// `newest_updated`, skip already-synced items, and collect the tasks -/// still needing ingest. Returns the queued items and whether we crossed -/// the persistent cursor boundary (the signal to stop paginating this -/// workspace). All order-dependent decisions (cursor/timestamp) live here -/// — never in the concurrent stage. -fn select_pending<'a>( - tasks: &'a [Value], - state: &SyncState, - newest_updated: &mut Option, -) -> (Vec>, bool) { - let mut hit_cursor_boundary = false; - let mut pending: Vec = Vec::new(); - for task in tasks { - let Some(task_id) = extract_item_id(task, TASK_ID_PATHS) else { - tracing::debug!("[composio:clickup] task missing ID, skipping"); - continue; - }; - - let updated = sync::extract_task_updated(task); - - // Track newest `date_updated` for cursor advancement. - if let Some(ref ts) = updated { - if newest_updated.as_ref().is_none_or(|existing| ts > existing) { - *newest_updated = Some(ts.clone()); - } - } - - // Composite (task_id, date_updated) key so a task edited *after* - // its last sync is re-persisted. Same trick Notion uses for - // `last_edited_time`. - let sync_key = match &updated { - Some(ts) => format!("{task_id}@{ts}"), - None => task_id.clone(), - }; - - // Older than cursor AND already synced → caught up. - if let (Some(ref cursor), Some(ref ts)) = (&state.cursor, &updated) { - if ts <= cursor && state.is_synced(&sync_key) { - hit_cursor_boundary = true; - continue; - } - } - - if state.is_synced(&sync_key) { - continue; - } - - let title_text = - sync::extract_task_name(task).unwrap_or_else(|| format!("ClickUp task {task_id}")); - let title = format!("ClickUp: {title_text}"); - - pending.push(PendingIngest { - sync_key, - task_id, - title, - updated, - task, - }); - } - (pending, hit_cursor_boundary) -} - -/// Pass 2: ingest the queued tasks with bounded concurrency. Overlaps the -/// per-item embedding RTT (`buffer_unordered`, up to `concurrency` in -/// flight) and folds results into an order-independent -/// [`BufferedIngestOutcome`]. Unordered is correct here: nothing -/// downstream depends on completion order — successes are keyed by -/// `sync_key`. A failed ingest is logged and skipped (parity with the -/// previous sequential `continue`); ClickUp advances its cursor regardless -/// of per-item failures. -async fn ingest_pending_buffered( - ingestor: &I, - pending: Vec>, - workspace_id: &str, - concurrency: usize, -) -> BufferedIngestOutcome { - // Materialize the per-item futures into a Vec before `buffer_unordered` - // so the spawned sync future keeps concrete lifetimes / `Send`. - let ingest_futs = pending - .into_iter() - .map(|p| async move { - let res = ingestor - .ingest(&p.task_id, &p.title, p.updated.as_deref(), p.task) - .await; - (p.sync_key, p.task_id, res) - }) - .collect::>(); - - let mut outcome = BufferedIngestOutcome::default(); - let mut ingest_stream = futures::stream::iter(ingest_futs).buffer_unordered(concurrency); - while let Some((sync_key, task_id, res)) = ingest_stream.next().await { - match res { - Ok(_chunks_written) => { - outcome.synced_keys.push(sync_key); - outcome.persisted += 1; - } - Err(e) => { - tracing::warn!( - task_id = %task_id, - workspace_id = %workspace_id, - error = %e, - "[composio:clickup] ingest failed (continuing)" - ); - } - } - } - outcome -} - -#[cfg(test)] -mod buffered_tests { - use super::*; - use serde_json::json; - use std::sync::atomic::{AtomicUsize, Ordering}; - use std::sync::Arc; - - /// Fake ingestor: records the peak number of concurrent in-flight - /// `ingest` calls and can be told to fail one specific `task_id`. No - /// memory tree or embedder involved — lets us assert the concurrency - /// bound and overlap deterministically. - struct CountingIngestor { - in_flight: AtomicUsize, - peak: AtomicUsize, - fail_task: Option, - } - - impl CountingIngestor { - fn new(fail_task: Option<&str>) -> Arc { - Arc::new(Self { - in_flight: AtomicUsize::new(0), - peak: AtomicUsize::new(0), - fail_task: fail_task.map(str::to_string), - }) - } - } - - #[async_trait] - impl TaskIngestor for CountingIngestor { - async fn ingest( - &self, - task_id: &str, - _title: &str, - _updated: Option<&str>, - _task: &Value, - ) -> anyhow::Result { - let now = self.in_flight.fetch_add(1, Ordering::SeqCst) + 1; - self.peak.fetch_max(now, Ordering::SeqCst); - // Yield a few times so futures genuinely interleave and the - // peak counter reflects real overlap, not accidental serial run. - for _ in 0..4 { - tokio::task::yield_now().await; - } - self.in_flight.fetch_sub(1, Ordering::SeqCst); - if self.fail_task.as_deref() == Some(task_id) { - Err(anyhow::anyhow!("forced failure for {task_id}")) - } else { - Ok(1) - } - } - } - - fn make_tasks(n: usize) -> Vec { - (0..n).map(|i| json!({ "id": format!("t{i}") })).collect() - } - - fn make_pending(tasks: &[Value]) -> Vec> { - tasks - .iter() - .enumerate() - .map(|(i, task)| PendingIngest { - sync_key: format!("k{i}"), - task_id: format!("t{i}"), - title: format!("ClickUp: task {i}"), - updated: None, - task, - }) - .collect() - } - - #[tokio::test] - async fn ingest_pending_buffered_bounds_and_overlaps() { - let ingestor = CountingIngestor::new(None); - let tasks = make_tasks(20); - let pending = make_pending(&tasks); - - let outcome = ingest_pending_buffered(ingestor.as_ref(), pending, "ws1", 8).await; - - assert_eq!(outcome.persisted, 20, "all tasks persisted"); - assert_eq!(outcome.synced_keys.len(), 20); - - let peak = ingestor.peak.load(Ordering::SeqCst); - assert!(peak <= 8, "peak in-flight {peak} exceeded the bound of 8"); - assert!(peak >= 2, "peak in-flight {peak} shows no real overlap"); - } - - #[tokio::test] - async fn ingest_pending_buffered_skips_failures_order_independent() { - let ingestor = CountingIngestor::new(Some("t2")); - let tasks = make_tasks(5); - let pending = make_pending(&tasks); - - let outcome = ingest_pending_buffered(ingestor.as_ref(), pending, "ws1", 4).await; - - assert_eq!(outcome.persisted, 4, "the one failed ingest is not counted"); - assert_eq!(outcome.synced_keys.len(), 4); - assert!( - !outcome.synced_keys.contains(&"k2".to_string()), - "the failed task's sync_key must not be marked synced" - ); - } - - #[test] - fn select_pending_tracks_newest_skips_synced_and_detects_boundary() { - let mut state = SyncState::new("clickup", "conn1"); - state.cursor = Some("2026-04-15T00:00:00Z".to_string()); - // Task B is already synced and older than the cursor. - state.mark_synced("b@2026-04-01T00:00:00Z"); - - let tasks = vec![ - json!({ "id": "a", "date_updated": "2026-05-01T00:00:00Z" }), - json!({ "id": "b", "date_updated": "2026-04-01T00:00:00Z" }), - json!({ "date_updated": "2026-03-01T00:00:00Z" }), // no id → skipped - ]; - - let mut newest: Option = None; - let (pending, hit_boundary) = select_pending(&tasks, &state, &mut newest); - - assert_eq!(pending.len(), 1, "only the new task A is queued"); - assert_eq!(pending[0].task_id, "a"); - assert_eq!(pending[0].sync_key, "a@2026-05-01T00:00:00Z"); - assert!( - hit_boundary, - "older synced task B trips the cursor boundary" - ); - assert_eq!(newest.as_deref(), Some("2026-05-01T00:00:00Z")); - } -} diff --git a/src/openhuman/memory_sync/composio/providers/clickup/source.rs b/src/openhuman/memory_sync/composio/providers/clickup/source.rs new file mode 100644 index 000000000..34f5d2c6b --- /dev/null +++ b/src/openhuman/memory_sync/composio/providers/clickup/source.rs @@ -0,0 +1,569 @@ +//! ClickUp's [`IncrementalSource`] primitives. +//! +//! ClickUp rides the generic +//! [`crate::openhuman::memory_sync::composio::providers::orchestrator`]: +//! [`ClickUpProvider::sync`](super::provider::ClickUpProvider) delegates to +//! [`run_clickup_sync`]. The orchestrator owns the control flow (budget, +//! pagination bound, dedup, the `sync_depth_days` window, the precise +//! `max_items` clamp, cursor advance/hold, state persistence); this module +//! supplies only the ClickUp-specific shapes. +//! +//! ClickUp is the first **nested** provider on the orchestrator: +//! [`ClickUpSource::preamble`] resolves the authorized user id + the visible +//! workspaces and returns **one [`SyncScope`] per workspace**, then the +//! orchestrator pages each workspace's `CLICKUP_GET_FILTERED_TEAM_TASKS` +//! (1-indexed `page`) scoped to that user as assignee. The shared user id is +//! resolved once in the preamble and stashed on the source ([`OnceLock`]) so +//! every per-workspace `fetch_page` can read it back. +//! +//! Two provider quirks the trait already accommodates: +//! * **epoch-ms timestamps** — `date_updated` is a millisecond-epoch string, +//! so [`ClickUpSource::depth_floor`] emits an epoch-ms floor (not RFC3339) +//! for the lexicographic compare to be valid. +//! * **advance-on-failure** — ClickUp advances its cursor even when a per-item +//! ingest fails, so it overrides +//! [`IncrementalSource::hold_cursor_on_ingest_failure`] to `false`. + +use std::sync::OnceLock; + +use async_trait::async_trait; +use futures::StreamExt; +use serde_json::{json, Value}; + +use super::provider::{ + ACTION_GET_AUTHORIZED_TEAMS_WORKSPACES, ACTION_GET_AUTHORIZED_USER, + ACTION_GET_FILTERED_TEAM_TASKS, TASK_ID_PATHS, +}; +use super::{ingest::ingest_task_into_memory_tree, sync}; +use crate::openhuman::config::Config; +use crate::openhuman::memory_sync::composio::providers::orchestrator::{ + self, IncrementalSource, IngestOutcome, PageFetch, SyncItem, SyncScope, +}; +use crate::openhuman::memory_sync::composio::providers::sync_state::{extract_item_id, SyncState}; +use crate::openhuman::memory_sync::composio::providers::{ + ProviderContext, SyncOutcome, SyncReason, +}; + +/// Page size per API call on steady-state syncs. +const PAGE_SIZE: u32 = 50; + +/// Larger initial-sync page size so the first backfill catches up faster. +const INITIAL_PAGE_SIZE: u32 = 100; + +/// Maximum pages **per workspace** per sync pass. +const MAX_PAGES_PER_WORKSPACE: u32 = 20; + +/// Max in-flight ingests per page. DB writes serialize anyway and the cloud +/// embedder has rate limits, so keep this small. +const INGEST_CONCURRENCY: usize = 8; + +/// ClickUp's [`IncrementalSource`]. Holds the authorized user id resolved in +/// the preamble so every per-workspace `fetch_page` can scope to it. +pub(crate) struct ClickUpSource { + user_id: OnceLock, +} + +impl ClickUpSource { + fn new() -> Self { + Self { + user_id: OnceLock::new(), + } + } +} + +/// Entry point used by [`super::provider::ClickUpProvider::sync`]. +pub(crate) async fn run_clickup_sync( + ctx: &ProviderContext, + reason: SyncReason, +) -> Result { + orchestrator::run_sync(&ClickUpSource::new(), ctx, reason).await +} + +impl ClickUpSource { + /// Look up (and budget-record) the authorized user's numeric id. + async fn resolve_user_id( + &self, + ctx: &ProviderContext, + state: &mut SyncState, + ) -> Result { + tracing::debug!( + connection_id = ?ctx.connection_id, + "[composio:clickup] resolve_user_id via {ACTION_GET_AUTHORIZED_USER}" + ); + + let resp = ctx + .execute(ACTION_GET_AUTHORIZED_USER, Some(json!({}))) + .await + .map_err(|e| { + format!("[composio:clickup] {ACTION_GET_AUTHORIZED_USER} failed: {e:#}") + })?; + state.record_requests(1); + + if !resp.successful { + let err = resp + .error + .clone() + .unwrap_or_else(|| "provider reported failure".to_string()); + return Err(format!( + "[composio:clickup] {ACTION_GET_AUTHORIZED_USER}: {err}" + )); + } + + let user_id = sync::extract_user_id(&resp.data).ok_or_else(|| { + "[composio:clickup] CLICKUP_GET_AUTHORIZED_USER returned no user.id".to_string() + })?; + + tracing::debug!( + connection_id = ?ctx.connection_id, + "[composio:clickup] resolve_user_id complete" + ); + Ok(user_id) + } + + /// Look up (and budget-record) the workspace (team) ids visible to this + /// connection. + async fn resolve_workspaces( + &self, + ctx: &ProviderContext, + state: &mut SyncState, + ) -> Result, String> { + tracing::debug!( + connection_id = ?ctx.connection_id, + "[composio:clickup] resolve_workspaces via {ACTION_GET_AUTHORIZED_TEAMS_WORKSPACES}" + ); + + let resp = ctx + .execute(ACTION_GET_AUTHORIZED_TEAMS_WORKSPACES, Some(json!({}))) + .await + .map_err(|e| { + format!("[composio:clickup] {ACTION_GET_AUTHORIZED_TEAMS_WORKSPACES} failed: {e:#}") + })?; + state.record_requests(1); + + if !resp.successful { + let err = resp + .error + .clone() + .unwrap_or_else(|| "provider reported failure".to_string()); + return Err(format!( + "[composio:clickup] {ACTION_GET_AUTHORIZED_TEAMS_WORKSPACES}: {err}" + )); + } + + let workspaces = sync::extract_workspace_ids(&resp.data); + tracing::debug!( + connection_id = ?ctx.connection_id, + count = workspaces.len(), + "[composio:clickup] resolve_workspaces complete" + ); + Ok(workspaces) + } +} + +#[async_trait] +impl IncrementalSource for ClickUpSource { + fn toolkit(&self) -> &'static str { + "clickup" + } + + fn page_size(&self, reason: SyncReason) -> u32 { + match reason { + SyncReason::ConnectionCreated => INITIAL_PAGE_SIZE, + _ => PAGE_SIZE, + } + } + + fn max_pages(&self) -> u32 { + MAX_PAGES_PER_WORKSPACE + } + + fn detail_noun(&self) -> &'static str { + "tasks" + } + + /// ClickUp advances its cursor even on per-item ingest failures. + fn hold_cursor_on_ingest_failure(&self) -> bool { + false + } + + /// `date_updated` is a millisecond-epoch string, so the depth floor must be + /// epoch-ms too (not RFC3339) for the lexicographic compare to hold. + fn depth_floor(&self, days: u32) -> String { + let floor = chrono::Utc::now() - chrono::Duration::days(days as i64); + floor.timestamp_millis().to_string() + } + + /// Resolve the user id (stashed for `fetch_page`) and return one scope per + /// visible workspace. Mirrors the original's budget re-check: if the daily + /// budget is exhausted right after the user-id probe, we skip the + /// workspaces call (empty scopes → no-op outcome). + async fn preamble( + &self, + ctx: &ProviderContext, + state: &mut SyncState, + ) -> Result, String> { + let user_id = self.resolve_user_id(ctx, state).await?; + let _ = self.user_id.set(user_id); + + if state.budget_exhausted() { + tracing::info!( + "[composio:clickup] budget exhausted after user-id probe, skipping sync" + ); + return Ok(vec![]); + } + + let workspaces = self.resolve_workspaces(ctx, state).await?; + Ok(workspaces + .into_iter() + .map(|ws| { + let label = format!("workspace:{ws}"); + SyncScope::nested(ws, label) + }) + .collect()) + } + + async fn fetch_page( + &self, + ctx: &ProviderContext, + scope: &SyncScope, + cursor: Option<&str>, + reason: SyncReason, + state: &mut SyncState, + ) -> Result { + let workspace_id = &scope.id; + let user_id = self.user_id.get().cloned().unwrap_or_default(); + // ClickUp paginates by 0-indexed `page`; the orchestrator's opaque + // cursor carries the next page number (`None` = first page). + let page_num: u32 = cursor.and_then(|c| c.parse().ok()).unwrap_or(0); + let page_size = self.page_size(reason); + + let args = json!({ + "team_id": workspace_id, + "assignees": [user_id], + "order_by": "updated", + "reverse": true, + "page": page_num, + "page_size": page_size, + "subtasks": true, + }); + + tracing::debug!( + connection_id = ?ctx.connection_id, + workspace_id = %workspace_id, + page_num, + page_size, + "[composio:clickup] fetch_page via {ACTION_GET_FILTERED_TEAM_TASKS}" + ); + + let resp = ctx + .execute(ACTION_GET_FILTERED_TEAM_TASKS, Some(args)) + .await + .map_err(|e| { + format!( + "[composio:clickup] {ACTION_GET_FILTERED_TEAM_TASKS} \ + workspace={workspace_id} page={page_num}: {e:#}" + ) + })?; + state.record_requests(1); + + if !resp.successful { + let err = resp + .error + .clone() + .unwrap_or_else(|| "provider reported failure".to_string()); + return Err(format!( + "[composio:clickup] {ACTION_GET_FILTERED_TEAM_TASKS} \ + workspace={workspace_id} page={page_num}: {err}" + )); + } + + let tasks = sync::extract_tasks(&resp.data); + // ClickUp signals the last page implicitly: a short page ends the + // workspace. + let next = if (tasks.len() as u32) < page_size { + None + } else { + Some((page_num + 1).to_string()) + }; + + tracing::debug!( + connection_id = ?ctx.connection_id, + workspace_id = %workspace_id, + page_num, + fetched = tasks.len(), + has_next = next.is_some(), + "[composio:clickup] fetch_page complete" + ); + + Ok(PageFetch { items: tasks, next }) + } + + fn item_dedup_key(&self, item: &Value) -> Option { + let task_id = extract_item_id(item, TASK_ID_PATHS)?; + match sync::extract_task_updated(item) { + Some(updated) => Some(format!("{task_id}@{updated}")), + None => Some(task_id), + } + } + + fn item_sort_ts(&self, item: &Value) -> Option { + sync::extract_task_updated(item) + } + + async fn ingest( + &self, + ctx: &ProviderContext, + scope: &SyncScope, + _state: &mut SyncState, + items: Vec, + ) -> IngestOutcome { + let connection_id = ctx.connection_id.as_deref().unwrap_or("default"); + + let pending: Vec = items + .into_iter() + .filter_map(|it| { + let task_id = extract_item_id(&it.raw, TASK_ID_PATHS)?; + let title_text = sync::extract_task_name(&it.raw) + .unwrap_or_else(|| format!("ClickUp task {task_id}")); + Some(PendingIngest { + sync_key: it.dedup_key, + task_id, + title: format!("ClickUp: {title_text}"), + updated: it.sort_ts, + task: it.raw, + }) + }) + .collect(); + + let ingestor = MemoryTreeIngestor { + config: ctx.config.as_ref(), + connection_id, + }; + let buffered = + ingest_pending_buffered(&ingestor, pending, &scope.id, INGEST_CONCURRENCY).await; + IngestOutcome { + synced_keys: buffered.synced_keys, + persisted: buffered.persisted, + had_failures: buffered.had_failures, + } + } +} + +/// One task queued for concurrent ingest. Owns its raw task `Value` (the +/// orchestrator handed ownership via [`SyncItem`]). +struct PendingIngest { + sync_key: String, + task_id: String, + title: String, + updated: Option, + task: Value, +} + +/// Folded result of [`ingest_pending_buffered`]. Order-independent. +#[derive(Default)] +struct BufferedIngestOutcome { + synced_keys: Vec, + persisted: usize, + had_failures: bool, +} + +/// Seam over "ingest one ClickUp task", so the bounded-concurrency driver can +/// be unit-tested with a fake that records peak in-flight calls. +#[async_trait] +trait TaskIngestor { + async fn ingest( + &self, + task_id: &str, + title: &str, + updated: Option<&str>, + task: &Value, + ) -> anyhow::Result; +} + +/// Production ingestor: routes into the memory-tree pipeline. +struct MemoryTreeIngestor<'c> { + config: &'c Config, + connection_id: &'c str, +} + +#[async_trait] +impl TaskIngestor for MemoryTreeIngestor<'_> { + async fn ingest( + &self, + task_id: &str, + title: &str, + updated: Option<&str>, + task: &Value, + ) -> anyhow::Result { + ingest_task_into_memory_tree( + self.config, + self.connection_id, + task_id, + title, + updated, + task, + ) + .await + } +} + +/// Ingest queued tasks with bounded concurrency, folding into an +/// order-independent [`BufferedIngestOutcome`]. A failed ingest is logged and +/// skipped (ClickUp advances its cursor regardless, via +/// `hold_cursor_on_ingest_failure = false`). +async fn ingest_pending_buffered( + ingestor: &I, + pending: Vec, + workspace_id: &str, + concurrency: usize, +) -> BufferedIngestOutcome { + let ingest_futs = pending + .into_iter() + .map(|p| async move { + let res = ingestor + .ingest(&p.task_id, &p.title, p.updated.as_deref(), &p.task) + .await; + (p.sync_key, p.task_id, res) + }) + .collect::>(); + + let mut outcome = BufferedIngestOutcome::default(); + let mut ingest_stream = futures::stream::iter(ingest_futs).buffer_unordered(concurrency); + while let Some((sync_key, task_id, res)) = ingest_stream.next().await { + match res { + Ok(_chunks_written) => { + outcome.synced_keys.push(sync_key); + outcome.persisted += 1; + } + Err(e) => { + outcome.had_failures = true; + tracing::warn!( + task_id = %task_id, + workspace_id = %workspace_id, + error = %e, + "[composio:clickup] ingest failed (continuing)" + ); + } + } + } + outcome +} + +#[cfg(test)] +mod buffered_tests { + use super::*; + use serde_json::json; + use std::sync::atomic::{AtomicUsize, Ordering}; + use std::sync::Arc; + + /// Fake ingestor: records peak concurrent in-flight `ingest` calls and can + /// fail one specific `task_id`. No memory tree or embedder involved. + struct CountingIngestor { + in_flight: AtomicUsize, + peak: AtomicUsize, + fail_task: Option, + } + + impl CountingIngestor { + fn new(fail_task: Option<&str>) -> Arc { + Arc::new(Self { + in_flight: AtomicUsize::new(0), + peak: AtomicUsize::new(0), + fail_task: fail_task.map(str::to_string), + }) + } + } + + #[async_trait] + impl TaskIngestor for CountingIngestor { + async fn ingest( + &self, + task_id: &str, + _title: &str, + _updated: Option<&str>, + _task: &Value, + ) -> anyhow::Result { + let now = self.in_flight.fetch_add(1, Ordering::SeqCst) + 1; + self.peak.fetch_max(now, Ordering::SeqCst); + for _ in 0..4 { + tokio::task::yield_now().await; + } + self.in_flight.fetch_sub(1, Ordering::SeqCst); + if self.fail_task.as_deref() == Some(task_id) { + Err(anyhow::anyhow!("forced failure for {task_id}")) + } else { + Ok(1) + } + } + } + + fn make_pending(n: usize) -> Vec { + (0..n) + .map(|i| PendingIngest { + sync_key: format!("k{i}"), + task_id: format!("t{i}"), + title: format!("ClickUp: task {i}"), + updated: None, + task: json!({ "id": format!("t{i}") }), + }) + .collect() + } + + #[tokio::test] + async fn ingest_pending_buffered_bounds_and_overlaps() { + let ingestor = CountingIngestor::new(None); + let pending = make_pending(20); + + let outcome = ingest_pending_buffered(ingestor.as_ref(), pending, "ws1", 8).await; + + assert_eq!(outcome.persisted, 20, "all tasks persisted"); + assert_eq!(outcome.synced_keys.len(), 20); + assert!(!outcome.had_failures); + + let peak = ingestor.peak.load(Ordering::SeqCst); + assert!(peak <= 8, "peak in-flight {peak} exceeded the bound of 8"); + assert!(peak >= 2, "peak in-flight {peak} shows no real overlap"); + } + + #[tokio::test] + async fn ingest_pending_buffered_skips_failures_order_independent() { + let ingestor = CountingIngestor::new(Some("t2")); + let pending = make_pending(5); + + let outcome = ingest_pending_buffered(ingestor.as_ref(), pending, "ws1", 4).await; + + assert_eq!(outcome.persisted, 4, "the one failed ingest is not counted"); + assert!(outcome.had_failures); + assert_eq!(outcome.synced_keys.len(), 4); + assert!(!outcome.synced_keys.contains(&"k2".to_string())); + } + + #[test] + fn item_dedup_key_composes_id_and_updated() { + let with_update = json!({ "id": "t1", "date_updated": "1780000000000" }); + assert_eq!( + ClickUpSource::new().item_dedup_key(&with_update).as_deref(), + Some("t1@1780000000000") + ); + let no_update = json!({ "id": "t2" }); + assert_eq!( + ClickUpSource::new().item_dedup_key(&no_update).as_deref(), + Some("t2") + ); + assert_eq!( + ClickUpSource::new().item_dedup_key(&json!({ "date_updated": "x" })), + None + ); + } + + #[test] + fn depth_floor_is_epoch_millis() { + let floor = ClickUpSource::new().depth_floor(7); + // Epoch-ms string: all digits, 13 chars in the 2020s+. + assert!(floor.chars().all(|c| c.is_ascii_digit()), "got {floor}"); + assert!( + floor.len() >= 13, + "epoch-ms should be >=13 digits, got {floor}" + ); + } +} diff --git a/tests/memory_sync_providers_raw_coverage_e2e.rs b/tests/memory_sync_providers_raw_coverage_e2e.rs index 13e9bc061..38c95e095 100644 --- a/tests/memory_sync_providers_raw_coverage_e2e.rs +++ b/tests/memory_sync_providers_raw_coverage_e2e.rs @@ -1489,6 +1489,87 @@ async fn clickup_sync_max_items_caps_ingest_to_exact_count() { server.abort(); } +// ───────────────────────────────────────────────────────────────────────── +// ClickUp sync_depth_days enforcement (via the shared orchestrator) +// +// ClickUp's `date_updated` is an epoch-millis string, so the orchestrator's +// depth floor must be epoch-ms too (ClickUpSource::depth_floor override). The +// mock returns 2 recent tasks (year-2030 epoch-ms) + 3 ancient (year-2001) in +// descending order; with sync_depth_days=7 only the 2 recent tasks persist. +// ───────────────────────────────────────────────────────────────────────── + +/// Build `recent` + `old` ClickUp tasks in descending `date_updated` order, +/// using epoch-millis strings so the epoch-ms depth floor compares correctly. +fn clickup_depth_tasks(recent: usize, old: usize) -> Vec { + let mut tasks = Vec::new(); + for i in 0..recent { + tasks.push(json!({ + "id": format!("clickup-recent-{i:04}"), + "name": format!("Recent {i}"), + "text_content": "recent", + "status": { "status": "open" }, + "date_updated": format!("{}", 1_900_000_000_000_u64 + (recent - i) as u64), + "url": format!("https://app.clickup.com/t/clickup-recent-{i:04}") + })); + } + for i in 0..old { + tasks.push(json!({ + "id": format!("clickup-old-{i:04}"), + "name": format!("Old {i}"), + "text_content": "old", + "status": { "status": "open" }, + "date_updated": format!("{}", 1_000_000_000_000_u64 + (old - i) as u64), + "url": format!("https://app.clickup.com/t/clickup-old-{i:04}") + })); + } + tasks +} + +#[tokio::test] +async fn clickup_sync_depth_days_filters_old_tasks() { + let _guard = env_lock(); + let tmp = TempDir::new().expect("tempdir"); + let _workspace = EnvGuard::set_path("OPENHUMAN_WORKSPACE", tmp.path()); + let _home = EnvGuard::set_path("HOME", tmp.path()); + let _backend = EnvGuard::unset("BACKEND_URL"); + + // One workspace, one page: 2 recent + 3 ancient. With a 7-day window only + // the 2 recent tasks must be ingested. + let requests: Arc>> = Arc::new(Mutex::new(Vec::new())); + let (base, server) = loopback_router(clickup_cap_router( + clickup_depth_tasks(2, 3), + Arc::clone(&requests), + )) + .await; + + let mut config = config_in(&tmp); + config.api_url = Some(base); + persist_config(&config).await; + store_session(&config); + memory_global::init(config.workspace_dir.clone()).expect("init global memory"); + + let ctx = ProviderContext { + config: Arc::new(config), + toolkit: "clickup".to_string(), + connection_id: Some("conn-clickup-depth".to_string()), + usage: Default::default(), + max_items: None, + sync_depth_days: Some(7), + }; + + let outcome = ClickUpProvider::new() + .sync(&ctx, SyncReason::ConnectionCreated) + .await + .expect("clickup depth sync"); + + assert_eq!( + outcome.items_ingested, 2, + "sync_depth_days=7 must drop the 3 year-2001 tasks and keep the 2 recent ones" + ); + + server.abort(); +} + fn walk_files(root: &Path) -> Vec { let mut out = Vec::new(); if !root.exists() {