admin/openhuman
1
0
Fork 0
mirror of https://github.com/tinyhumansai/openhuman.git synced 2026-07-27 21:08:00 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(memory_sync): convert ClickUp to the generic sync orchestrator (Refs #3333, stacked on #3860) (#3873)

Browse Source 
Co-authored-by: Claude <noreply@anthropic.com>
Co-authored-by: Steven Enamakel <enamakel@tinyhumans.ai>
This commit is contained in:
sanil-23
2026-06-21 17:27:56 -07:00
committed by GitHub
co-authored by Claude Steven Enamakel
parent 3e0244670c
commit 2791638818
4 changed files with 662 additions and 727 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/openhuman/memory_sync/composio/providers/clickup/mod.rs
+1
View File
@@ -15,6 +15,7 @@
mod ingest;
mod provider;
mod source;
mod sync;
#[cfg(test)]
mod tests;
src/openhuman/memory_sync/composio/providers/clickup/provider.rs
+11 -727
View File
@@ -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<SyncOutcome, String> {
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<String> = 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<String> = 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<NormalizedTask> {
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<String, String> {
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<Vec<String>, 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<String>,
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<String>,
/// 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<usize>;
}
/// 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<usize> {
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<String>,
) -> (Vec<PendingIngest<'a>>, bool) {
let mut hit_cursor_boundary = false;
let mut pending: Vec<PendingIngest> = 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<I: TaskIngestor + Sync>(
ingestor: &I,
pending: Vec<PendingIngest<'_>>,
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::<Vec<_>>();
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<String>,
}
impl CountingIngestor {
fn new(fail_task: Option<&str>) -> Arc<Self> {
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<usize> {
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<Value> {
(0..n).map(|i| json!({ "id": format!("t{i}") })).collect()
}
fn make_pending(tasks: &[Value]) -> Vec<PendingIngest<'_>> {
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<String> = 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"));
}
}
src/openhuman/memory_sync/composio/providers/clickup/source.rs
+569
View File
@@ -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<String>,
}
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<SyncOutcome, String> {
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<String, String> {
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<Vec<String>, 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<Vec<SyncScope>, 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<PageFetch, String> {
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<String> {
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<String> {
sync::extract_task_updated(item)
}
async fn ingest(
&self,
ctx: &ProviderContext,
scope: &SyncScope,
_state: &mut SyncState,
items: Vec<SyncItem>,
) -> IngestOutcome {
let connection_id = ctx.connection_id.as_deref().unwrap_or("default");
let pending: Vec<PendingIngest> = 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<String>,
task: Value,
}
/// Folded result of [`ingest_pending_buffered`]. Order-independent.
#[derive(Default)]
struct BufferedIngestOutcome {
synced_keys: Vec<String>,
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<usize>;
}
/// 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<usize> {
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<I: TaskIngestor + Sync>(
ingestor: &I,
pending: Vec<PendingIngest>,
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::<Vec<_>>();
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<String>,
}
impl CountingIngestor {
fn new(fail_task: Option<&str>) -> Arc<Self> {
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<usize> {
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<PendingIngest> {
(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}"
);
}
}
tests/memory_sync_providers_raw_coverage_e2e.rs
+81
View File
@@ -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<Value> {
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<Mutex<Vec<Value>>> = 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<std::path::PathBuf> {
let mut out = Vec::new();
if !root.exists() {