From 56a31c88bdbf443ecb22cdc4bedc179ddf0e4b26 Mon Sep 17 00:00:00 2001 From: YOMXXX Date: Tue, 2 Jun 2026 14:04:30 +0800 Subject: [PATCH] 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 --- app/scripts/e2e-web-session.sh | 5 + .../OpenhumanLinkModal.notifications.test.tsx | 4 +- .../intelligence/memoryGraphLayout.test.ts | 2 +- app/test/playwright/helpers/core-rpc.ts | 12 +- .../specs/chat-harness-wallet-flow.spec.ts | 8 +- .../specs/harness-search-tool-flow.spec.ts | 2 +- .../settings-feature-preferences.spec.ts | 54 ++-- src/openhuman/accessibility/focus.rs | 137 +++++++-- .../memory_sources/readers/github.rs | 32 +- .../composio/providers/clickup/ingest.rs | 290 ++++++++++++++++++ .../composio/providers/clickup/mod.rs | 4 +- .../composio/providers/clickup/provider.rs | 20 +- src/openhuman/memory_tools/types.rs | 7 + .../config_auth_app_state_connectivity_e2e.rs | 31 +- 14 files changed, 525 insertions(+), 83 deletions(-) create mode 100644 src/openhuman/memory_sync/composio/providers/clickup/ingest.rs diff --git a/app/scripts/e2e-web-session.sh b/app/scripts/e2e-web-session.sh index bc7835b12..3e1488c30 100755 --- a/app/scripts/e2e-web-session.sh +++ b/app/scripts/e2e-web-session.sh @@ -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" \ diff --git a/app/src/components/__tests__/OpenhumanLinkModal.notifications.test.tsx b/app/src/components/__tests__/OpenhumanLinkModal.notifications.test.tsx index a5b5a4110..aaf7dab39 100644 --- a/app/src/components/__tests__/OpenhumanLinkModal.notifications.test.tsx +++ b/app/src/components/__tests__/OpenhumanLinkModal.notifications.test.tsx @@ -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); diff --git a/app/src/components/intelligence/memoryGraphLayout.test.ts b/app/src/components/intelligence/memoryGraphLayout.test.ts index 628fdaacf..299576783 100644 --- a/app/src/components/intelligence/memoryGraphLayout.test.ts +++ b/app/src/components/intelligence/memoryGraphLayout.test.ts @@ -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); diff --git a/app/test/playwright/helpers/core-rpc.ts b/app/test/playwright/helpers/core-rpc.ts index 6d9745a6b..43fdaac54 100644 --- a/app/test/playwright/helpers/core-rpc.ts +++ b/app/test/playwright/helpers/core-rpc.ts @@ -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 { async function completeAuthCallback(page: Page, token: string): Promise { 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 { 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/); } diff --git a/app/test/playwright/specs/chat-harness-wallet-flow.spec.ts b/app/test/playwright/specs/chat-harness-wallet-flow.spec.ts index 02d6f8be6..3b15d42f8 100644 --- a/app/test/playwright/specs/chat-harness-wallet-flow.spec.ts +++ b/app/test/playwright/specs/chat-harness-wallet-flow.spec.ts @@ -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 }); }); }); diff --git a/app/test/playwright/specs/harness-search-tool-flow.spec.ts b/app/test/playwright/specs/harness-search-tool-flow.spec.ts index 9b2e647ed..f52c279df 100644 --- a/app/test/playwright/specs/harness-search-tool-flow.spec.ts +++ b/app/test/playwright/specs/harness-search-tool-flow.spec.ts @@ -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(); diff --git a/app/test/playwright/specs/settings-feature-preferences.spec.ts b/app/test/playwright/specs/settings-feature-preferences.spec.ts index f3a035508..ebf4cfbd5 100644 --- a/app/test/playwright/specs/settings-feature-preferences.spec.ts +++ b/app/test/playwright/specs/settings-feature-preferences.spec.ts @@ -52,11 +52,40 @@ async function getMascotVoiceId(page: Page): Promise { }); } +async function getPersistedMascotColor(page: Page): Promise { + 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 { 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('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('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('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'); diff --git a/src/openhuman/accessibility/focus.rs b/src/openhuman/accessibility/focus.rs index 7a84b9ca6..cbb5cf096 100644 --- a/src/openhuman/accessibility/focus.rs +++ b/src/openhuman/accessibility/focus.rs @@ -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 { + 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 { 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 { 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 { 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")); + } +} diff --git a/src/openhuman/memory_sources/readers/github.rs b/src/openhuman/memory_sources/readers/github.rs index 853ea14ea..a1c44b3d2 100644 --- a/src/openhuman/memory_sources/readers/github.rs +++ b/src/openhuman/memory_sources/readers/github.rs @@ -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), diff --git a/src/openhuman/memory_sync/composio/providers/clickup/ingest.rs b/src/openhuman/memory_sync/composio/providers/clickup/ingest.rs new file mode 100644 index 000000000..d01dab342 --- /dev/null +++ b/src/openhuman/memory_sync/composio/providers/clickup/ingest.rs @@ -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 { + raw.and_then(|s| s.trim().parse::().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 { + 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 { + 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 = 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" + ); + } +} diff --git a/src/openhuman/memory_sync/composio/providers/clickup/mod.rs b/src/openhuman/memory_sync/composio/providers/clickup/mod.rs index e749ba3fe..5bb012498 100644 --- a/src/openhuman/memory_sync/composio/providers/clickup/mod.rs +++ b/src/openhuman/memory_sync/composio/providers/clickup/mod.rs @@ -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)] diff --git a/src/openhuman/memory_sync/composio/providers/clickup/provider.rs b/src/openhuman/memory_sync/composio/providers/clickup/provider.rs index b237ec5e8..75d7e9a79 100644 --- a/src/openhuman/memory_sync/composio/providers/clickup/provider.rs +++ b/src/openhuman/memory_sync/composio/providers/clickup/provider.rs @@ -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)" ); } } diff --git a/src/openhuman/memory_tools/types.rs b/src/openhuman/memory_tools/types.rs index e512faa1f..d2a659ecb 100644 --- a/src/openhuman/memory_tools/types.rs +++ b/src/openhuman/memory_tools/types.rs @@ -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] diff --git a/tests/config_auth_app_state_connectivity_e2e.rs b/tests/config_auth_app_state_connectivity_e2e.rs index 0198d2159..6ecef3d10 100644 --- a/tests/config_auth_app_state_connectivity_e2e.rs +++ b/tests/config_auth_app_state_connectivity_e2e.rs @@ -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");