diff --git a/app/test/core-rpc-playwright-helper.test.ts b/app/test/core-rpc-playwright-helper.test.ts new file mode 100644 index 000000000..98dd17feb --- /dev/null +++ b/app/test/core-rpc-playwright-helper.test.ts @@ -0,0 +1,25 @@ +import type { Page } from '@playwright/test'; +import { describe, expect, it } from 'vitest'; + +import { seedBrowserCoreMode } from './playwright/helpers/core-rpc'; + +describe('Playwright core RPC helper', () => { + it('primes walkthrough completion before the app renders', async () => { + localStorage.clear(); + + const page = { + async addInitScript( + script: (args: { rpcUrl: string; token: string }) => void, + args: { rpcUrl: string; token: string } + ) { + script(args); + }, + } as unknown as Page; + + await seedBrowserCoreMode(page); + + expect(localStorage.getItem('openhuman_core_mode')).toBe('cloud'); + expect(localStorage.getItem('openhuman:walkthrough_completed')).toBe('true'); + expect(localStorage.getItem('openhuman:walkthrough_pending')).toBeNull(); + }); +}); diff --git a/app/test/playwright/helpers/core-rpc.ts b/app/test/playwright/helpers/core-rpc.ts index 1af112ce2..07c274025 100644 --- a/app/test/playwright/helpers/core-rpc.ts +++ b/app/test/playwright/helpers/core-rpc.ts @@ -53,6 +53,8 @@ export async function seedBrowserCoreMode(page: Page): Promise { window.localStorage.setItem('openhuman_core_mode', 'cloud'); window.localStorage.setItem('openhuman_core_rpc_url', rpcUrl); window.localStorage.setItem('openhuman_core_rpc_token', token); + window.localStorage.setItem('openhuman:walkthrough_completed', 'true'); + window.localStorage.removeItem('openhuman:walkthrough_pending'); }, { rpcUrl: CORE_RPC_URL, token: CORE_RPC_TOKEN } ); @@ -64,6 +66,8 @@ async function applyBrowserCoreModeInPage(page: Page): Promise { window.localStorage.setItem('openhuman_core_mode', 'cloud'); window.localStorage.setItem('openhuman_core_rpc_url', rpcUrl); window.localStorage.setItem('openhuman_core_rpc_token', token); + window.localStorage.setItem('openhuman:walkthrough_completed', 'true'); + window.localStorage.removeItem('openhuman:walkthrough_pending'); }, { rpcUrl: CORE_RPC_URL, token: CORE_RPC_TOKEN } ); diff --git a/src/openhuman/memory_sync/composio/providers/linear/ingest.rs b/src/openhuman/memory_sync/composio/providers/linear/ingest.rs new file mode 100644 index 000000000..9438c8e38 --- /dev/null +++ b/src/openhuman/memory_sync/composio/providers/linear/ingest.rs @@ -0,0 +1,273 @@ +//! Linear -> memory tree ingest plumbing. +//! +//! Converts one Linear issue 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. + +use anyhow::Result; +use chrono::{DateTime, 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 Linear document metadata. +pub const LINEAR_PLATFORM: &str = "linear"; + +/// Stable tags attached to every Linear-ingested issue chunk. +pub const DEFAULT_TAGS: &[&str] = &["linear", "ingested"]; + +/// Build the memory-tree source id for one Linear issue in one connection. +pub(crate) fn linear_source_id(connection_id: &str, issue_id: &str) -> String { + format!("linear:{connection_id}:{issue_id}") +} + +/// Render the raw Linear issue payload as a markdown document body. +fn render_issue_body(title: &str, issue: &Value) -> String { + let pretty = serde_json::to_string_pretty(issue).unwrap_or_else(|_| "{}".to_string()); + format!("# {title}\n\n```json\n{pretty}\n```\n") +} + +/// Parse Linear's `updatedAt` timestamp, falling back to now on malformed input. +fn parse_updated_time(raw: Option<&str>) -> DateTime { + raw.and_then(|s| DateTime::parse_from_rfc3339(s).ok()) + .map(|dt| dt.with_timezone(&Utc)) + .unwrap_or_else(Utc::now) +} + +/// Ingest one Linear issue into memory_tree and return the written chunk count. +/// +/// Edited issues 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_issue_into_memory_tree( + config: &Config, + connection_id: &str, + issue_id: &str, + title: &str, + updated_time: Option<&str>, + issue: &Value, +) -> Result { + let source_id = linear_source_id(connection_id, issue_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, + issue_id = %issue_id, + removed_chunks = removed, + "[composio:linear] ingest: re-ingest cleanup" + ); + } + + let modified_at = parse_updated_time(updated_time); + let body = render_issue_body(title, issue); + let source_ref = Some(format!("linear://issue/{issue_id}")); + let doc = DocumentInput { + provider: LINEAR_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!("linear:{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, + issue_id = %issue_id, + chunks_written, + already_ingested, + "[composio:linear] ingest: issue 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::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_issue(issue_id: &str, updated_at: &str) -> Value { + json!({ + "id": issue_id, + "identifier": "ENG-42", + "title": "Fix external LLM routing", + "updatedAt": updated_at, + "url": "https://linear.app/openhuman/issue/ENG-42/fix-external-llm-routing", + "description": "Connected app tools vanish under external routing.", + "state": { "name": "In Progress" }, + "team": { "key": "ENG", "name": "Engineering" }, + "assignee": { "name": "Alice" } + }) + } + + #[test] + fn linear_source_id_is_stable_and_namespaced() { + let a = linear_source_id("conn-1", "issue-abc"); + let b = linear_source_id("conn-1", "issue-abc"); + assert_eq!(a, b); + assert_eq!(a, "linear:conn-1:issue-abc"); + assert_ne!(a, linear_source_id("conn-2", "issue-abc")); + assert_ne!(a, linear_source_id("conn-1", "issue-xyz")); + } + + #[test] + fn parse_updated_time_handles_valid_and_invalid_inputs() { + let good = parse_updated_time(Some("2026-05-28T12:34:56.000Z")); + assert_eq!(good.format("%Y-%m-%d").to_string(), "2026-05-28"); + + let bad = parse_updated_time(Some("not-a-timestamp")); + assert!((Utc::now() - bad).num_seconds().abs() < 5); + + let missing = parse_updated_time(None); + assert!((Utc::now() - missing).num_seconds().abs() < 5); + } + + #[test] + fn render_issue_body_includes_title_header_and_pretty_json() { + let issue = json!({ + "id": "issue-1", + "identifier": "ENG-42", + "title": "Fix external LLM routing" + }); + let body = render_issue_body("Linear: Fix external LLM routing", &issue); + assert!(body.starts_with("# Linear: Fix external LLM routing\n")); + assert!(body.contains("```json\n")); + assert!(body.contains("\"identifier\": \"ENG-42\"")); + assert!(body.contains("\"title\": \"Fix external LLM routing\"")); + } + + #[tokio::test] + async fn ingest_issue_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-linear"; + let issue_id = "issue-routing"; + let issue = sample_issue(issue_id, "2026-05-28T10:00:00.000Z"); + let chunks_before = count_chunks(&cfg).expect("count_chunks before"); + + let written = ingest_issue_into_memory_tree( + &cfg, + connection_id, + issue_id, + "Linear: Fix external LLM routing", + Some("2026-05-28T10:00:00.000Z"), + &issue, + ) + .await + .expect("ingest_issue_into_memory_tree"); + + assert!(written > 0, "Linear 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 = linear_source_id(connection_id, issue_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_issue_replaces_prior_chunks() { + use crate::openhuman::memory_store::chunks::store::count_chunks; + + let (_tmp, cfg) = test_config(); + let connection_id = "conn-edit"; + let issue_id = "issue-edit"; + + let v1 = sample_issue(issue_id, "2026-05-28T10:00:00.000Z"); + let first = ingest_issue_into_memory_tree( + &cfg, + connection_id, + issue_id, + "Linear: Fix external LLM routing", + Some("2026-05-28T10:00:00.000Z"), + &v1, + ) + .await + .expect("first ingest"); + assert!(first > 0); + let after_first = count_chunks(&cfg).expect("count after first"); + + let v2 = json!({ + "id": issue_id, + "identifier": "ENG-42", + "title": "Fix external LLM routing", + "updatedAt": "2026-05-29T10:00:00.000Z", + "description": "Updated: external LLM routing now keeps connected app tools visible.", + "state": { "name": "Done" } + }); + let second = ingest_issue_into_memory_tree( + &cfg, + connection_id, + issue_id, + "Linear: Fix external LLM routing", + Some("2026-05-29T10:00:00.000Z"), + &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 issue must replace prior chunks, not append duplicates" + ); + } +} diff --git a/src/openhuman/memory_sync/composio/providers/linear/mod.rs b/src/openhuman/memory_sync/composio/providers/linear/mod.rs index d744dbe6a..bb40292b8 100644 --- a/src/openhuman/memory_sync/composio/providers/linear/mod.rs +++ b/src/openhuman/memory_sync/composio/providers/linear/mod.rs @@ -3,6 +3,7 @@ //! //! Issue: #2400. +mod ingest; mod provider; mod sync; #[cfg(test)] diff --git a/src/openhuman/memory_sync/composio/providers/linear/provider.rs b/src/openhuman/memory_sync/composio/providers/linear/provider.rs index af9385a16..51ac5ff91 100644 --- a/src/openhuman/memory_sync/composio/providers/linear/provider.rs +++ b/src/openhuman/memory_sync/composio/providers/linear/provider.rs @@ -1,5 +1,5 @@ //! Linear provider — incremental sync of issues assigned to the -//! authenticated user, with per-item persistence into the Memory Tree. +//! authenticated user, with per-issue memory_tree ingest. //! //! On each sync pass: //! @@ -9,8 +9,8 @@ //! 4. Page through `LINEAR_LIST_LINEAR_ISSUES` filtered to the viewer as //! assignee, ordered by `updatedAt` descending. Stop early once we hit //! issues older than the cursor or a page without a next-page cursor. -//! 5. For each issue, persist as a single memory document if it's new -//! *or* edited since the last sync. +//! 5. For each issue, ingest into memory_tree if it's new *or* edited +//! since the last sync. //! 6. Advance the cursor to the newest `updatedAt` seen and save. //! //! Privacy posture: we only pull issues the user is assigned to, never @@ -21,10 +21,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_issue_into_memory_tree, sync}; +use crate::openhuman::memory_sync::composio::providers::sync_state::{extract_item_id, SyncState}; use crate::openhuman::memory_sync::composio::providers::{ merge_extra, pick_str, ComposioProvider, CuratedTool, NormalizedTask, ProviderContext, ProviderUserProfile, SyncOutcome, SyncReason, TaskFetchFilter, @@ -252,12 +250,7 @@ impl ComposioProvider for LinearProvider { // ── Per-item dedup + persist ───────────────────────────── for issue in &issues { - let Some(issue_id) = - crate::openhuman::memory_sync::composio::providers::sync_state::extract_item_id( - issue, - ISSUE_ID_PATHS, - ) - else { + let Some(issue_id) = extract_item_id(issue, ISSUE_ID_PATHS) else { tracing::debug!("[composio:linear] issue missing ID, skipping"); continue; }; @@ -294,17 +287,15 @@ impl ComposioProvider for LinearProvider { let title_text = sync::extract_issue_title(issue) .unwrap_or_else(|| format!("Linear issue {issue_id}")); - let doc_id = format!("composio-linear-issue-{issue_id}"); let title = format!("Linear: {title_text}"); - match persist_single_item( - &memory, - "linear", - &doc_id, + match ingest_issue_into_memory_tree( + &ctx.config, + &connection_id, + &issue_id, &title, + updated.as_deref(), issue, - "linear", - ctx.connection_id.as_deref(), ) .await { @@ -317,7 +308,7 @@ impl ComposioProvider for LinearProvider { tracing::warn!( issue_id = %issue_id, error = %e, - "[composio:linear] failed to persist issue (continuing)" + "[composio:linear] failed to ingest issue into memory_tree (continuing)" ); } } @@ -481,11 +472,7 @@ impl ComposioProvider for LinearProvider { /// Map a raw Linear issue payload into a [`NormalizedTask`]. fn normalize_linear_issue(issue: &serde_json::Value) -> Option { - let external_id = - crate::openhuman::memory_sync::composio::providers::sync_state::extract_item_id( - issue, - ISSUE_ID_PATHS, - )?; + let external_id = extract_item_id(issue, ISSUE_ID_PATHS)?; let title = sync::extract_issue_title(issue).unwrap_or_else(|| format!("Linear issue {external_id}")); Some(NormalizedTask { diff --git a/src/openhuman/memory_tree/retrieval/source.rs b/src/openhuman/memory_tree/retrieval/source.rs index 9c0852165..c910d3d89 100644 --- a/src/openhuman/memory_tree/retrieval/source.rs +++ b/src/openhuman/memory_tree/retrieval/source.rs @@ -267,6 +267,7 @@ const PLATFORM_KINDS: &[(&str, &str)] = &[ ("protonmail", "email"), // Document platforms ("notion", "document"), + ("linear", "document"), ("drive", "document"), ("googledoc", "document"), ("doc", "document"), @@ -502,6 +503,7 @@ mod tests { assert!(scope_matches_kind("slack:#eng", "chat")); assert!(scope_matches_kind("gmail:alice", "email")); assert!(scope_matches_kind("notion:page123", "document")); + assert!(scope_matches_kind("linear:conn-1:issue-abc", "document")); assert!(!scope_matches_kind("slack:#eng", "email")); assert!(scope_matches_kind("chat:custom", "chat")); }