fix(memory-sync): migrate ClickUp sync to memory_tree pipeline (#2885) (#3080)

Co-authored-by: Cyrus Gray <144336577+graycyrus@users.noreply.github.com> Co-authored-by: Steven Enamakel <enamakel@tinyhumans.ai>
2026-07-27 21:08:00 +00:00 · 2026-06-01 23:04:30 -07:00
parent 13e2af1095
commit 56a31c88bd
14 changed files with 525 additions and 83 deletions
@@ -99,6 +99,9 @@ reasoning_provider = "e2e:e2e-mock-model"
 agentic_provider = "e2e:e2e-mock-model"
 coding_provider = "e2e:e2e-mock-model"

+[update]
+enabled = false
+
 [[cloud_providers]]
 id = "p_e2e_mock"
 slug = "e2e"
@@ -120,6 +123,8 @@ fi

 export OPENHUMAN_CORE_TOKEN="$PW_CORE_RPC_TOKEN"
 export OPENHUMAN_TELEGRAM_BOT_API_BASE="http://127.0.0.1:${E2E_MOCK_PORT}"
+# Keep the standalone core aligned with the Rust mock runner: sub-agent
+# orchestration builds large async futures and can overflow the default stack.
 export RUST_MIN_STACK="${RUST_MIN_STACK:-16777216}"

 "$OPENHUMAN_CORE_BIN" run --host 127.0.0.1 --port "$OPENHUMAN_CORE_PORT" \
@@ -46,7 +46,7 @@ describe('OpenhumanLinkModal notifications test flow', () => {

    await waitFor(() => {
      expect(
-        screen.getByText(/Test notification sent\. If you didn't receive it/i)
+        screen.getByText(/Test notification sent\. If you didn['’]t receive it/i)
      ).toBeInTheDocument();
    });
    expect(showNativeNotification).toHaveBeenCalledWith(
@@ -116,7 +116,7 @@ describe('OpenhumanLinkModal notifications test flow', () => {

    await waitFor(() => {
      expect(
-        screen.getByText(/Test notification sent\. If you didn't receive it/i)
+        screen.getByText(/Test notification sent\. If you didn['’]t receive it/i)
      ).toBeInTheDocument();
    });
    expect(showNativeNotification).toHaveBeenCalledTimes(1);
@@ -41,7 +41,7 @@ describe('memoryGraphLayout', () => {
    expect(nodeColor(contact())).toBe(CONTACT_COLOR);
  });

-  it('nodeRadius grows with summary level and is fixed for chunk/contact', () => {
+  it('nodeRadius grows with level and is fixed for chunk/contact', () => {
    expect(nodeRadius(summary({ level: 0 }))).toBe(5);
    expect(nodeRadius(summary({ level: 3 }))).toBe(12.5);
    expect(nodeRadius(summary({ level: 99 }))).toBe(252.5);
@@ -2,6 +2,7 @@ import { expect, type Page } from '@playwright/test';

 const CORE_RPC_URL = process.env.PW_CORE_RPC_URL || 'http://127.0.0.1:17788/rpc';
 const CORE_RPC_TOKEN = process.env.PW_CORE_RPC_TOKEN || 'openhuman-playwright-token';
+const AUTH_CALLBACK_HOME_TIMEOUT_MS = 30_000;

 let nextRpcId = 1;

@@ -76,8 +77,13 @@ async function applyBrowserCoreModeInPage(page: Page): Promise<void> {
 async function completeAuthCallback(page: Page, token: string): Promise<void> {
  await page.goto(`/#/callback/auth?token=${encodeURIComponent(token)}&key=auth`);
  try {
+    // The app-side auth callback waits up to 15s for CoreStateProvider to
+    // commit currentUser before navigating to /home; CI occasionally needs
+    // more than Playwright's default 10s assertion window here.
    await expect
-      .poll(async () => page.evaluate(() => window.location.hash), { timeout: 10_000 })
+      .poll(async () => page.evaluate(() => window.location.hash), {
+        timeout: AUTH_CALLBACK_HOME_TIMEOUT_MS,
+      })
      .toMatch(/^#\/home/);
    return;
  } catch {
@@ -96,7 +102,9 @@ async function completeAuthCallback(page: Page, token: string): Promise<void> {
  await applyBrowserCoreModeInPage(page);
  await page.goto(`/#/callback/auth?token=${encodeURIComponent(token)}&key=auth`);
  await expect
-    .poll(async () => page.evaluate(() => window.location.hash), { timeout: 15_000 })
+    .poll(async () => page.evaluate(() => window.location.hash), {
+      timeout: AUTH_CALLBACK_HOME_TIMEOUT_MS,
+    })
    .toMatch(/^#\/home/);
 }

@@ -194,9 +194,11 @@ test.describe('Chat Harness - Wallet Flow', () => {
    await sendMessage(page, WALLET_PROMPT);

    await expect(
-      page.getByText(
-        /Prepared a wallet quote for John\..*wallet-quote-canary-8d13|Done\.\s*wallet-quote-canary-8d13/i
-      )
+      page
+        .getByText(
+          /Prepared a wallet quote for John\..*wallet-quote-canary-8d13|Done\.\s*wallet-quote-canary-8d13/i
+        )
+        .first()
    ).toBeVisible({ timeout: 40_000 });
  });
 });
@@ -186,7 +186,7 @@ test.describe('Harness - Search tool-flow', () => {
    await setMockBehavior('llmStreamChunkDelayMs', '10');

    await sendMessage(page, 'search for Rust async best practices');
-    await expect(page.getByText(CANARY)).toBeVisible({ timeout: 60_000 });
+    await expect(page.getByText(CANARY).first()).toBeVisible({ timeout: 60_000 });
    await expect(
      page.getByText(/Here are the top results for Rust async best practices/i)
    ).toBeVisible();
@@ -52,11 +52,40 @@ async function getMascotVoiceId(page: Page): Promise<string | null> {
  });
 }

+async function getPersistedMascotColor(page: Page): Promise<string | null> {
+  return page.evaluate(() => {
+    const userId = localStorage.getItem('OPENHUMAN_ACTIVE_USER_ID');
+    if (!userId) return null;
+
+    const raw = localStorage.getItem(`${userId}:persist:mascot`);
+    if (!raw) return null;
+
+    try {
+      const parsed = JSON.parse(raw) as { color?: unknown };
+      if (typeof parsed.color !== 'string') return null;
+      const color = JSON.parse(parsed.color) as unknown;
+      return typeof color === 'string' ? color : null;
+    } catch {
+      return null;
+    }
+  });
+}
+
 async function getAriaChecked(page: Page, label: string): Promise<string | null> {
  const value = await page.getByRole('switch', { name: label }).getAttribute('aria-checked');
  return value;
 }

+interface ToolsSnapshot {
+  result?: { localState?: { onboardingTasks?: { enabledTools?: string[] | null } | null } | null };
+  localState?: { onboardingTasks?: { enabledTools?: string[] | null } | null } | null;
+}
+
+function readEnabledTools(snapshot: ToolsSnapshot): string[] {
+  const body = snapshot.result ?? snapshot;
+  return body.localState?.onboardingTasks?.enabledTools ?? [];
+}
+
 test.describe('Settings - Feature Preferences', () => {
  test('renders the features settings section route', async ({ page }) => {
    await openAuthenticatedRoute(page, 'pw-settings-features-route', '/settings/features');
@@ -100,12 +129,8 @@ test.describe('Settings - Feature Preferences', () => {
      },
    });

-    const before = await callCoreRpc<{
-      result?: {
-        localState?: { onboardingTasks?: { enabledTools?: string[] | null } | null } | null;
-      };
-    }>('openhuman.app_state_snapshot', {});
-    const enabledBefore = before.result?.localState?.onboardingTasks?.enabledTools ?? [];
+    const before = await callCoreRpc<ToolsSnapshot>('openhuman.app_state_snapshot', {});
+    const enabledBefore = readEnabledTools(before);

    await reloadAndWait(page);

@@ -122,22 +147,14 @@ test.describe('Settings - Feature Preferences', () => {

    await expect
      .poll(async () => {
-        const after = await callCoreRpc<{
-          result?: {
-            localState?: { onboardingTasks?: { enabledTools?: string[] | null } | null } | null;
-          };
-        }>('openhuman.app_state_snapshot', {});
-        const enabledAfter = after.result?.localState?.onboardingTasks?.enabledTools ?? [];
+        const after = await callCoreRpc<ToolsSnapshot>('openhuman.app_state_snapshot', {});
+        const enabledAfter = readEnabledTools(after);
        return JSON.stringify(enabledAfter) !== JSON.stringify(enabledBefore);
      })
      .toBe(true);

-    const after = await callCoreRpc<{
-      result?: {
-        localState?: { onboardingTasks?: { enabledTools?: string[] | null } | null } | null;
-      };
-    }>('openhuman.app_state_snapshot', {});
-    expect(after.result?.localState?.onboardingTasks?.enabledTools ?? []).not.toContain('shell');
+    const after = await callCoreRpc<ToolsSnapshot>('openhuman.app_state_snapshot', {});
+    expect(readEnabledTools(after)).not.toContain('shell');
  });

  test('persists notifications DND and category preferences', async ({ page }) => {
@@ -178,6 +195,7 @@ test.describe('Settings - Feature Preferences', () => {
    await expect(page.getByRole('heading', { name: 'Color', exact: true })).toBeVisible();
    await page.getByTestId('mascot-color-burgundy').click();
    await expect(page.getByTestId('mascot-color-burgundy')).toHaveAttribute('aria-checked', 'true');
+    await expect.poll(() => getPersistedMascotColor(page)).toBe('burgundy');

    await reloadAndWait(page);
    await expect(page.getByTestId('mascot-color-burgundy')).toHaveAttribute('aria-checked', 'true');
@@ -8,6 +8,53 @@ use super::terminal::{is_terminal_app, is_text_role};
 #[cfg(target_os = "macos")]
 use super::text_util::{normalize_ax_value, parse_ax_number};
 use super::types::{AppContext, ElementBounds, FocusedTextContext};
+#[cfg(any(target_os = "macos", test))]
+use std::{
+    process::{Command, Output, Stdio},
+    time::{Duration, Instant},
+};
+
+#[cfg(target_os = "macos")]
+const FOREGROUND_CONTEXT_COMMAND_TIMEOUT: Duration = Duration::from_millis(1_500);
+#[cfg(any(target_os = "macos", test))]
+const COMMAND_TIMEOUT_POLL_INTERVAL: Duration = Duration::from_millis(10);
+
+#[cfg(any(target_os = "macos", test))]
+fn command_output_with_timeout(
+    command_name: &str,
+    command: &mut Command,
+    timeout: Duration,
+) -> Result<Output, String> {
+    command.stdout(Stdio::piped()).stderr(Stdio::piped());
+    let mut child = command
+        .spawn()
+        .map_err(|e| format!("failed to run {command_name}: {e}"))?;
+    let started_at = Instant::now();
+
+    loop {
+        match child.try_wait() {
+            Ok(Some(_)) => {
+                return child
+                    .wait_with_output()
+                    .map_err(|e| format!("failed to collect {command_name} output: {e}"));
+            }
+            Ok(None) if started_at.elapsed() >= timeout => {
+                let _ = child.kill();
+                let _ = child.wait();
+                return Err(format!(
+                    "{command_name} timed out after {}ms",
+                    timeout.as_millis()
+                ));
+            }
+            Ok(None) => std::thread::sleep(COMMAND_TIMEOUT_POLL_INTERVAL),
+            Err(error) => {
+                let _ = child.kill();
+                let _ = child.wait();
+                return Err(format!("failed to wait for {command_name}: {error}"));
+            }
+        }
+    }
+}

 // ---------------------------------------------------------------------------
 // Focus query: unified helper → osascript fallback
@@ -264,11 +311,13 @@ fn focused_text_via_osascript() -> Result<FocusedTextContext, String> {
      end tell
    "##;

-    let output = std::process::Command::new("osascript")
-        .arg("-e")
-        .arg(script)
-        .output()
-        .map_err(|e| format!("failed to run osascript: {e}"))?;
+    let mut command = Command::new("osascript");
+    command.arg("-e").arg(script);
+    let output = command_output_with_timeout(
+        "osascript focused_text_via_osascript",
+        &mut command,
+        FOREGROUND_CONTEXT_COMMAND_TIMEOUT,
+    )?;
    if !output.status.success() {
        let stderr = String::from_utf8_lossy(&output.stderr).trim().to_string();
        if stderr.is_empty() {
@@ -471,11 +520,19 @@ pub fn foreground_context() -> Option<AppContext> {
      end tell
    "#;

-    let output = std::process::Command::new("osascript")
-        .arg("-e")
-        .arg(script)
-        .output()
-        .ok()?;
+    let mut command = Command::new("osascript");
+    command.arg("-e").arg(script);
+    let output = match command_output_with_timeout(
+        "osascript foreground_context",
+        &mut command,
+        FOREGROUND_CONTEXT_COMMAND_TIMEOUT,
+    ) {
+        Ok(output) => output,
+        Err(error) => {
+            tracing::debug!("[accessibility] foreground_context failed: {error}");
+            return None;
+        }
+    };

    if !output.status.success() {
        let stderr = String::from_utf8_lossy(&output.stderr);
@@ -632,14 +689,23 @@ if fallback > 0 {{ print(fallback) }}
        has_title_swift = if has_title { "true" } else { "false" },
    );

-    // Note: this subprocess has no explicit timeout. This is consistent with
-    // the rest of the codebase (`screencapture`, `osascript`) which also run
-    // without timeouts. Swift startup for a trivial snippet is typically <1s.
-    let output = std::process::Command::new("swift")
-        .arg("-e")
-        .arg(&swift_code)
-        .output()
-        .ok()?;
+    let mut command = Command::new("swift");
+    command.arg("-e").arg(&swift_code);
+    let output = match command_output_with_timeout(
+        "swift CGWindowList",
+        &mut command,
+        FOREGROUND_CONTEXT_COMMAND_TIMEOUT,
+    ) {
+        Ok(output) => output,
+        Err(error) => {
+            tracing::debug!(
+                "[accessibility] swift CGWindowList failed: {} app={:?}",
+                error,
+                app_name,
+            );
+            return None;
+        }
+    };

    if !output.status.success() {
        tracing::debug!(
@@ -674,3 +740,38 @@ pub fn validate_focused_target(
 pub fn foreground_context() -> Option<AppContext> {
    None
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[cfg(unix)]
+    #[test]
+    fn command_output_with_timeout_returns_output_for_fast_command() {
+        let mut command = Command::new("sh");
+        command.arg("-c").arg("printf ready");
+
+        let output =
+            command_output_with_timeout("test fast command", &mut command, Duration::from_secs(1))
+                .expect("fast command should complete");
+
+        assert!(output.status.success());
+        assert_eq!(String::from_utf8_lossy(&output.stdout), "ready");
+    }
+
+    #[cfg(unix)]
+    #[test]
+    fn command_output_with_timeout_kills_slow_command() {
+        let mut command = Command::new("sh");
+        command.arg("-c").arg("sleep 2; printf late");
+
+        let error = command_output_with_timeout(
+            "test slow command",
+            &mut command,
+            Duration::from_millis(50),
+        )
+        .expect_err("slow command should time out");
+
+        assert!(error.contains("timed out after"));
+    }
+}
@@ -869,10 +869,15 @@ async fn read_issue(
    let labels: Vec<&str> = issue.labels.iter().map(|l| l.name.as_str()).collect();
    let issue_body = issue.body.as_deref().unwrap_or("");

+    let comments = fetch_issue_comments(owner, repo, number, use_gh).await;
+    let participants =
+        unique_handles(std::iter::once(author).chain(comments.iter().map(|c| c.user.as_str())));
+
    let mut body = format!(
        "# Issue #{number}: {title}\n\n\
         **State:** {state}\n\
         **Author:** @{author}\n\
+         **Participants:** {participants}\n\
         **Labels:** {label_str}\n\
         **Created:** {created}\n\
         **Updated:** {updated}\n\n\
@@ -888,15 +893,13 @@ async fn read_issue(
        created = issue.created_at.as_deref().unwrap_or("unknown"),
        updated = issue.updated_at.as_deref().unwrap_or("unknown"),
    );
-    let comments = fetch_issue_comments(owner, repo, number, use_gh).await;
+
    if !comments.is_empty() {
        body.push_str("\n\n## Comments\n");
-        for comment in comments {
+        for comment in &comments {
            body.push_str(&format!(
-                "\n### @{user} at {created_at}\n\n{body}\n",
-                user = comment.user,
-                created_at = comment.created_at,
-                body = comment.body,
+                "\n### @{} ({})\n\n{}\n",
+                comment.user, comment.created_at, comment.body
            ));
        }
    }
@@ -949,10 +952,15 @@ async fn read_pr(
        None => "not merged".to_string(),
    };

-    let body = format!(
+    let comments = fetch_issue_comments(owner, repo, number, use_gh).await;
+    let participants =
+        unique_handles(std::iter::once(author).chain(comments.iter().map(|c| c.user.as_str())));
+
+    let mut body = format!(
        "# PR #{number}: {title}\n\n\
         **State:** {state} ({merged})\n\
         **Author:** @{author}\n\
+         **Participants:** {participants}\n\
         **Labels:** {label_str}\n\
         **Created:** {created}\n\
         **Updated:** {updated}\n\n\
@@ -970,6 +978,16 @@ async fn read_pr(
        updated = pr.updated_at.as_deref().unwrap_or("unknown"),
    );

+    if !comments.is_empty() {
+        body.push_str("\n\n## Comments\n");
+        for comment in &comments {
+            body.push_str(&format!(
+                "\n### @{} ({})\n\n{}\n",
+                comment.user, comment.created_at, comment.body
+            ));
+        }
+    }
+
    Ok(SourceContent {
        id: format!("pr:{number}"),
        title: format!("PR #{number} {}", pr.title),
@@ -0,0 +1,290 @@
+//! ClickUp -> memory tree ingest plumbing.
+//!
+//! Converts one ClickUp task payload into a memory_tree [`DocumentInput`]
+//! and calls `ingest_document` so retrieval surfaces read the content from
+//! `mem_tree_chunks` instead of the legacy `memory_docs` path (issue #2885).
+//!
+//! ClickUp differs from the Linear/Notion providers in one way: its
+//! `date_updated` field is a Unix epoch **milliseconds** value rendered as a
+//! string (e.g. `"1779962400000"`), not an RFC3339 timestamp — so
+//! [`parse_updated_time`] parses milliseconds rather than calling
+//! `DateTime::parse_from_rfc3339`.
+
+use anyhow::Result;
+use chrono::{DateTime, TimeZone, Utc};
+use serde_json::Value;
+
+use crate::openhuman::config::Config;
+use crate::openhuman::memory::ingest_pipeline::{self, IngestResult};
+use crate::openhuman::memory_store::chunks::store::{delete_chunks_by_source, is_source_ingested};
+use crate::openhuman::memory_store::chunks::types::SourceKind;
+use crate::openhuman::memory_sync::canonicalize::document::DocumentInput;
+
+/// Platform identifier embedded in ClickUp document metadata.
+pub const CLICKUP_PLATFORM: &str = "clickup";
+
+/// Stable tags attached to every ClickUp-ingested task chunk.
+pub const DEFAULT_TAGS: &[&str] = &["clickup", "ingested"];
+
+/// Build the memory-tree source id for one ClickUp task in one connection.
+pub(crate) fn clickup_source_id(connection_id: &str, task_id: &str) -> String {
+    format!("clickup:{connection_id}:{task_id}")
+}
+
+/// Render the raw ClickUp task payload as a markdown document body.
+fn render_task_body(title: &str, task: &Value) -> String {
+    let pretty = serde_json::to_string_pretty(task).unwrap_or_else(|_| "{}".to_string());
+    format!("# {title}\n\n```json\n{pretty}\n```\n")
+}
+
+/// Parse ClickUp's `date_updated` (Unix epoch **milliseconds** as a string),
+/// falling back to now on missing/malformed input.
+fn parse_updated_time(raw: Option<&str>) -> DateTime<Utc> {
+    raw.and_then(|s| s.trim().parse::<i64>().ok())
+        .and_then(|ms| Utc.timestamp_millis_opt(ms).single())
+        .unwrap_or_else(Utc::now)
+}
+
+/// Ingest one ClickUp task into memory_tree and return the written chunk count.
+///
+/// Edited tasks reuse the same `source_id`, so prior chunks are deleted before
+/// re-ingest to avoid the document pipeline's duplicate-source short-circuit.
+pub async fn ingest_task_into_memory_tree(
+    config: &Config,
+    connection_id: &str,
+    task_id: &str,
+    title: &str,
+    updated_time: Option<&str>,
+    task: &Value,
+) -> Result<usize> {
+    let source_id = clickup_source_id(connection_id, task_id);
+
+    let cfg_for_blocking = config.clone();
+    let source_for_blocking = source_id.clone();
+    let removed = tokio::task::spawn_blocking(move || -> Result<usize> {
+        if is_source_ingested(
+            &cfg_for_blocking,
+            SourceKind::Document,
+            &source_for_blocking,
+        )? {
+            delete_chunks_by_source(
+                &cfg_for_blocking,
+                SourceKind::Document,
+                &source_for_blocking,
+            )
+        } else {
+            Ok(0)
+        }
+    })
+    .await
+    .map_err(|e| anyhow::anyhow!("delete-prior task join error: {e}"))??;
+
+    if removed > 0 {
+        tracing::debug!(
+            connection_id = %connection_id,
+            task_id = %task_id,
+            removed_chunks = removed,
+            "[composio:clickup] ingest: re-ingest cleanup"
+        );
+    }
+
+    let modified_at = parse_updated_time(updated_time);
+    let body = render_task_body(title, task);
+    let source_ref = Some(format!("clickup://task/{task_id}"));
+    let doc = DocumentInput {
+        provider: CLICKUP_PLATFORM.to_string(),
+        title: title.to_string(),
+        body,
+        modified_at,
+        source_ref,
+    };
+    let tags: Vec<String> = DEFAULT_TAGS.iter().map(|s| s.to_string()).collect();
+    let owner = format!("clickup:{connection_id}");
+
+    match ingest_pipeline::ingest_document(config, &source_id, &owner, tags, doc).await {
+        Ok(IngestResult {
+            chunks_written,
+            already_ingested,
+            ..
+        }) => {
+            tracing::debug!(
+                connection_id = %connection_id,
+                task_id = %task_id,
+                chunks_written,
+                already_ingested,
+                "[composio:clickup] ingest: task persisted"
+            );
+            Ok(chunks_written)
+        }
+        Err(err) => Err(anyhow::anyhow!(
+            "ingest_document failed for {source_id}: {err:#}"
+        )),
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::openhuman::config::Config;
+    use crate::openhuman::memory_store::chunks::types::SourceKind;
+    use chrono::{TimeZone, Utc};
+    use serde_json::{json, Value};
+    use tempfile::TempDir;
+
+    fn test_config() -> (TempDir, Config) {
+        let tmp = TempDir::new().expect("tempdir");
+        let mut cfg = Config::default();
+        cfg.workspace_dir = tmp.path().to_path_buf();
+        cfg.memory_tree.embedding_endpoint = None;
+        cfg.memory_tree.embedding_model = None;
+        cfg.memory_tree.embedding_strict = false;
+        (tmp, cfg)
+    }
+
+    fn sample_task(task_id: &str, date_updated: &str) -> Value {
+        json!({
+            "id": task_id,
+            "name": "Fix external LLM routing",
+            "date_updated": date_updated,
+            "url": "https://app.clickup.com/t/abc123",
+            "status": { "status": "in progress" },
+            "text_content": "Connected app tools vanish under external routing.",
+            "assignees": [{ "username": "Alice" }]
+        })
+    }
+
+    #[test]
+    fn clickup_source_id_is_stable_and_namespaced() {
+        let a = clickup_source_id("conn-1", "task-abc");
+        let b = clickup_source_id("conn-1", "task-abc");
+        assert_eq!(a, b);
+        assert_eq!(a, "clickup:conn-1:task-abc");
+        assert_ne!(a, clickup_source_id("conn-2", "task-abc"));
+        assert_ne!(a, clickup_source_id("conn-1", "task-xyz"));
+    }
+
+    #[test]
+    fn parse_updated_time_handles_epoch_millis_and_invalid_inputs() {
+        // ClickUp sends epoch milliseconds as a string. Build the expected
+        // instant and round-trip through its millisecond representation so we
+        // never hand-compute the magic number.
+        let expected = Utc.with_ymd_and_hms(2026, 5, 28, 10, 0, 0).unwrap();
+        let ms = expected.timestamp_millis().to_string();
+        let good = parse_updated_time(Some(&ms));
+        assert_eq!(good, expected);
+
+        // Leading/trailing whitespace must still parse.
+        let padded = parse_updated_time(Some(&format!("  {ms}  ")));
+        assert_eq!(padded, expected);
+
+        let bad = parse_updated_time(Some("not-a-timestamp"));
+        assert!((Utc::now() - bad).num_seconds().abs() < 5);
+
+        // An RFC3339 string is *not* valid ClickUp input — it must fall back.
+        let rfc = parse_updated_time(Some("2026-05-28T10:00:00.000Z"));
+        assert!((Utc::now() - rfc).num_seconds().abs() < 5);
+
+        let missing = parse_updated_time(None);
+        assert!((Utc::now() - missing).num_seconds().abs() < 5);
+    }
+
+    #[test]
+    fn render_task_body_includes_title_header_and_pretty_json() {
+        let task = json!({
+            "id": "task-1",
+            "name": "Fix external LLM routing",
+            "date_updated": "1779962400000"
+        });
+        let body = render_task_body("ClickUp: Fix external LLM routing", &task);
+        assert!(body.starts_with("# ClickUp: Fix external LLM routing\n"));
+        assert!(body.contains("```json\n"));
+        assert!(body.contains("\"name\": \"Fix external LLM routing\""));
+        assert!(body.contains("\"date_updated\": \"1779962400000\""));
+    }
+
+    #[tokio::test]
+    async fn ingest_task_writes_to_memory_tree() {
+        use crate::openhuman::memory_store::chunks::store::{count_chunks, is_source_ingested};
+
+        let (_tmp, cfg) = test_config();
+        let connection_id = "conn-clickup";
+        let task_id = "task-routing";
+        let task = sample_task(task_id, "1779962400000");
+        let chunks_before = count_chunks(&cfg).expect("count_chunks before");
+
+        let written = ingest_task_into_memory_tree(
+            &cfg,
+            connection_id,
+            task_id,
+            "ClickUp: Fix external LLM routing",
+            Some("1779962400000"),
+            &task,
+        )
+        .await
+        .expect("ingest_task_into_memory_tree");
+
+        assert!(written > 0, "ClickUp ingest must write chunks");
+        let chunks_after = count_chunks(&cfg).expect("count_chunks after");
+        assert!(
+            chunks_after > chunks_before,
+            "ingest must populate mem_tree_chunks (#2885)"
+        );
+
+        let cfg_for_blocking = cfg.clone();
+        let expected = clickup_source_id(connection_id, task_id);
+        let registered = tokio::task::spawn_blocking(move || {
+            is_source_ingested(&cfg_for_blocking, SourceKind::Document, &expected).unwrap_or(false)
+        })
+        .await
+        .expect("source-check task join");
+        assert!(registered, "source_id must be registered");
+    }
+
+    #[tokio::test]
+    async fn re_ingesting_edited_task_replaces_prior_chunks() {
+        use crate::openhuman::memory_store::chunks::store::count_chunks;
+
+        let (_tmp, cfg) = test_config();
+        let connection_id = "conn-edit";
+        let task_id = "task-edit";
+
+        let v1 = sample_task(task_id, "1779962400000");
+        let first = ingest_task_into_memory_tree(
+            &cfg,
+            connection_id,
+            task_id,
+            "ClickUp: Fix external LLM routing",
+            Some("1779962400000"),
+            &v1,
+        )
+        .await
+        .expect("first ingest");
+        assert!(first > 0);
+        let after_first = count_chunks(&cfg).expect("count after first");
+
+        let v2 = json!({
+            "id": task_id,
+            "name": "Fix external LLM routing",
+            "date_updated": "1780048800000",
+            "text_content": "Updated: external LLM routing now keeps connected app tools visible.",
+            "status": { "status": "complete" }
+        });
+        let second = ingest_task_into_memory_tree(
+            &cfg,
+            connection_id,
+            task_id,
+            "ClickUp: Fix external LLM routing",
+            Some("1780048800000"),
+            &v2,
+        )
+        .await
+        .expect("second ingest");
+        assert!(second > 0);
+        let after_second = count_chunks(&cfg).expect("count after second");
+
+        assert!(
+            after_second.abs_diff(after_first) <= 1,
+            "edited task must replace prior chunks, not append duplicates"
+        );
+    }
+}
@@ -7,11 +7,13 @@
 //!
 //! - `provider.rs` — `impl ComposioProvider for ClickUpProvider`
 //! - `sync.rs`     — payload-shape helpers (results extraction, title)
+//! - `ingest.rs`   — memory_tree document ingest (issue #2885)
 //! - `tools.rs`    — `CLICKUP_CURATED` whitelist of Composio actions
 //! - `tests.rs`    — unit tests for the helpers + trait metadata
 //!
-//! Issue: #2288.
+//! Issue: #2288 (introduction); #2885 (memory_tree migration).

+mod ingest;
 mod provider;
 mod sync;
 #[cfg(test)]
@@ -27,10 +27,8 @@
 use async_trait::async_trait;
 use serde_json::json;

-use super::sync;
-use crate::openhuman::memory_sync::composio::providers::sync_state::{
-    persist_single_item, SyncState,
-};
+use super::{ingest::ingest_task_into_memory_tree, sync};
+use crate::openhuman::memory_sync::composio::providers::sync_state::SyncState;
 use crate::openhuman::memory_sync::composio::providers::{
    first_array_str, merge_extra, pick_str, ComposioProvider, CuratedTool, NormalizedTask,
    ProviderContext, ProviderUserProfile, SyncOutcome, SyncReason, TaskFetchFilter,
@@ -369,17 +367,15 @@ impl ComposioProvider for ClickUpProvider {

                    let title_text = sync::extract_task_name(task)
                        .unwrap_or_else(|| format!("ClickUp task {task_id}"));
-                    let doc_id = format!("composio-clickup-task-{task_id}");
                    let title = format!("ClickUp: {title_text}");

-                    match persist_single_item(
-                        &memory,
-                        "clickup",
-                        &doc_id,
+                    match ingest_task_into_memory_tree(
+                        &ctx.config,
+                        &connection_id,
+                        &task_id,
                        &title,
+                        updated.as_deref(),
                        task,
-                        "clickup",
-                        ctx.connection_id.as_deref(),
                    )
                    .await
                    {
@@ -392,7 +388,7 @@ impl ComposioProvider for ClickUpProvider {
                                task_id = %task_id,
                                workspace_id = %workspace_id,
                                error = %e,
-                                "[composio:clickup] failed to persist task (continuing)"
+                                "[composio:clickup] ingest failed (continuing)"
                            );
                        }
                    }
@@ -211,6 +211,13 @@ mod tests {
        let a = ToolMemoryRule::generate_id();
        let b = ToolMemoryRule::generate_id();
        assert_ne!(a, b);
+        assert!(a.starts_with('r'));
+        assert!(a[1..].chars().all(|c| matches!(c, 'a'..='p')));
+        assert!(
+            !crate::openhuman::memory_store::safety::pii::has_likely_pii(
+                &ToolMemoryRule::storage_key(&a)
+            )
+        );
    }

    #[test]
@@ -5463,32 +5463,27 @@ fn credentials_profile_store_recovers_dropped_entries_empty_files_and_datetime_e
        .to_string(),
    )
    .expect("write missing oauth secret fixture");
-    // An OAuth profile missing its access_token must not poison the whole
-    // store: it is dropped just like a bad-kind entry (see #3125), so the load
-    // succeeds with that single profile purged rather than returning an error.
-    let recovered_missing_secret = AuthProfilesStore::new(&missing_oauth_secret_dir, false)
+    let missing_secret = AuthProfilesStore::new(&missing_oauth_secret_dir, false)
        .load()
-        .expect("oauth profile missing access_token should be dropped, not fail the whole load");
-    assert!(
-        !recovered_missing_secret
-            .profiles
-            .contains_key("github:missing-access"),
-        "oauth profile missing access_token should be dropped on load: {recovered_missing_secret:#?}"
-    );
-    assert!(
-        !recovered_missing_secret.active_profiles.contains_key("github"),
-        "active profile pointing at a dropped profile should be purged: {recovered_missing_secret:#?}"
-    );
+        .expect("oauth profile missing access token should be dropped");
+    assert!(missing_secret.profiles.is_empty());
+    assert!(missing_secret.active_profiles.is_empty());
    let rewritten_missing_secret: Value = serde_json::from_str(
        &std::fs::read_to_string(missing_oauth_secret_dir.join("auth-profiles.json"))
-            .expect("read rewritten missing-oauth-secret store"),
+            .expect("read rewritten missing oauth secret profile store"),
    )
-    .expect("rewritten missing-oauth-secret store should be json");
+    .expect("rewritten missing oauth secret store should be json");
    assert!(
        rewritten_missing_secret
            .pointer("/profiles/github:missing-access")
            .is_none(),
-        "dropped oauth profile should be purged from persisted store: {rewritten_missing_secret}"
+        "missing oauth secret profile should be purged from persisted store: {rewritten_missing_secret}"
+    );
+    assert!(
+        rewritten_missing_secret
+            .pointer("/active_profiles/github")
+            .is_none(),
+        "active pointer to missing oauth secret profile should be purged: {rewritten_missing_secret}"
    );

    let public_api_dir = tmp.path().join("public-api-errors");