diff --git a/app/src/components/intelligence/AddMemorySourceDialog.test.tsx b/app/src/components/intelligence/AddMemorySourceDialog.test.tsx index 786deb461..57e3353d4 100644 --- a/app/src/components/intelligence/AddMemorySourceDialog.test.tsx +++ b/app/src/components/intelligence/AddMemorySourceDialog.test.tsx @@ -23,6 +23,7 @@ vi.mock('../../services/memorySourcesService', () => ({ SOURCE_KIND_ICONS: { folder: '📁', composio: '🔗', + conversation: '💬', github_repo: '🐙', rss_feed: '📡', web_page: '🌐', @@ -31,6 +32,7 @@ vi.mock('../../services/memorySourcesService', () => ({ SOURCE_KIND_LABEL_KEYS: { folder: 'memorySources.kind.folder', composio: 'memorySources.kind.composio', + conversation: 'memorySources.kind.conversation', github_repo: 'memorySources.kind.github_repo', rss_feed: 'memorySources.kind.rss_feed', web_page: 'memorySources.kind.web_page', @@ -202,6 +204,47 @@ describe('deduplicateConnections', () => { }); }); +// --------------------------------------------------------------------------- +// Component tests: Conversation kind +// --------------------------------------------------------------------------- + +describe('AddMemorySourceDialog — Conversation kind', () => { + beforeEach(() => { + mockListConnections.mockReset(); + mockGetSupportedToolkits.mockReset(); + mockGetSupportedToolkits.mockResolvedValue(DEFAULT_SUPPORTED); + }); + + it('shows no extra fields and submits with just a label', async () => { + const { addMemorySource } = await import('../../services/memorySourcesService'); + const mockAdd = addMemorySource as ReturnType; + mockAdd.mockResolvedValue({ + id: 'src_conv', + kind: 'conversation', + label: 'Chats', + enabled: true, + }); + + const { onAdded } = renderDialog(); + + const conversationBtn = screen.getByText('Conversation'); + fireEvent.click(conversationBtn); + + const labelInput = screen.getByPlaceholderText('My research notes'); + fireEvent.change(labelInput, { target: { value: 'Chats' } }); + + const submitBtn = screen.getByRole('button', { name: 'Add' }); + fireEvent.click(submitBtn); + + await waitFor(() => { + expect(mockAdd).toHaveBeenCalledWith( + expect.objectContaining({ kind: 'conversation', label: 'Chats', enabled: true }) + ); + }); + await waitFor(() => expect(onAdded).toHaveBeenCalled()); + }); +}); + // --------------------------------------------------------------------------- // Component tests: ComposioPicker inside the dialog // --------------------------------------------------------------------------- diff --git a/app/src/components/intelligence/AddMemorySourceDialog.tsx b/app/src/components/intelligence/AddMemorySourceDialog.tsx index d17c60dc8..10578c482 100644 --- a/app/src/components/intelligence/AddMemorySourceDialog.tsx +++ b/app/src/components/intelligence/AddMemorySourceDialog.tsx @@ -45,6 +45,7 @@ interface AddMemorySourceDialogProps { const ALL_KINDS: SourceKind[] = [ 'composio', + 'conversation', 'folder', 'github_repo', 'rss_feed', @@ -144,6 +145,8 @@ export function AddMemorySourceDialog({ open, onClose, onAdded }: AddMemorySourc params.toolkit = toolkit; params.connection_id = connectionId; break; + case 'conversation': + break; case 'folder': params.path = path.trim(); params.glob = glob.trim() || '**/*.md'; @@ -318,6 +321,8 @@ function isKindFieldsValid( switch (kind) { case 'composio': return fields.connectionId.length > 0; + case 'conversation': + return true; case 'folder': return fields.path.trim().length > 0; case 'github_repo': @@ -446,6 +451,8 @@ function KindFields(props: KindFieldsProps) { switch (props.kind) { case 'composio': return ; + case 'conversation': + return null; case 'folder': return ( <> diff --git a/app/src/components/intelligence/SourceSettingsPanel.tsx b/app/src/components/intelligence/SourceSettingsPanel.tsx index 24eeb9b5e..0a956ba56 100644 --- a/app/src/components/intelligence/SourceSettingsPanel.tsx +++ b/app/src/components/intelligence/SourceSettingsPanel.tsx @@ -24,6 +24,7 @@ import { // are intentionally omitted here. const KIND_FIELDS: Record> = { composio: ['sync_depth_days', 'max_items'], + conversation: ['sync_depth_days'], github_repo: ['max_prs', 'max_issues', 'max_commits', 'sync_depth_days'], rss_feed: ['max_items', 'sync_depth_days'], twitter_query: ['since_days'], diff --git a/app/src/lib/i18n/ar.ts b/app/src/lib/i18n/ar.ts index ea4c65570..0af4146c0 100644 --- a/app/src/lib/i18n/ar.ts +++ b/app/src/lib/i18n/ar.ts @@ -1986,6 +1986,7 @@ const messages: TranslationMap = { 'memorySources.cssSelectorPlaceholder': 'article', 'memorySources.searchQueryPlaceholder': 'من:', 'memorySources.kind.composio': 'التكامل', + 'memorySources.kind.conversation': 'محادثة', 'memorySources.kind.folder': 'محلي', 'memorySources.kind.github_repo': 'Xqx0x', 'memorySources.kind.twitter_query': 'Xqx0x', diff --git a/app/src/lib/i18n/bn.ts b/app/src/lib/i18n/bn.ts index f6bcd32a4..26c190ac4 100644 --- a/app/src/lib/i18n/bn.ts +++ b/app/src/lib/i18n/bn.ts @@ -2031,6 +2031,7 @@ const messages: TranslationMap = { 'memorySources.cssSelectorPlaceholder': 'article', 'memorySources.searchQueryPlaceholder': 'অজানা সংযোগ', 'memorySources.kind.composio': 'অবধি', + 'memorySources.kind.conversation': 'কথোপকথন', 'memorySources.kind.folder': 'স্থানীয় ফোল্ডার', 'memorySources.kind.github_repo': 'xqxqxRox', 'memorySources.kind.twitter_query': 'Xqxqx অনুসন্ধান', diff --git a/app/src/lib/i18n/de.ts b/app/src/lib/i18n/de.ts index 11471f76e..3af87b15d 100644 --- a/app/src/lib/i18n/de.ts +++ b/app/src/lib/i18n/de.ts @@ -2082,6 +2082,7 @@ const messages: TranslationMap = { 'memorySources.cssSelectorPlaceholder': 'article', 'memorySources.searchQueryPlaceholder': 'von:Benutzer AI Safety', 'memorySources.kind.composio': 'Integration', + 'memorySources.kind.conversation': 'Unterhaltung', 'memorySources.kind.folder': 'Lokaler Ordner', 'memorySources.kind.github_repo': 'GitHub-Repo', 'memorySources.kind.twitter_query': 'Twitter-Suche', diff --git a/app/src/lib/i18n/en.ts b/app/src/lib/i18n/en.ts index ad75b7648..e903e3b30 100644 --- a/app/src/lib/i18n/en.ts +++ b/app/src/lib/i18n/en.ts @@ -2489,6 +2489,7 @@ const en: TranslationMap = { 'memorySources.cssSelectorPlaceholder': 'article', 'memorySources.searchQueryPlaceholder': 'from:user AI safety', 'memorySources.kind.composio': 'Integration', + 'memorySources.kind.conversation': 'Conversation', 'memorySources.kind.folder': 'Local Folder', 'memorySources.kind.github_repo': 'GitHub Repo', 'memorySources.kind.twitter_query': 'Twitter Search', diff --git a/app/src/lib/i18n/es.ts b/app/src/lib/i18n/es.ts index 1f77e14bf..bb464b4e3 100644 --- a/app/src/lib/i18n/es.ts +++ b/app/src/lib/i18n/es.ts @@ -2073,6 +2073,7 @@ const messages: TranslationMap = { 'memorySources.cssSelectorPlaceholder': 'article', 'memorySources.searchQueryPlaceholder': 'de:usuario AI seguridad', 'memorySources.kind.composio': 'Integración', + 'memorySources.kind.conversation': 'Conversación', 'memorySources.kind.folder': 'Carpeta local', 'memorySources.kind.github_repo': 'GitHub Repositorio', 'memorySources.kind.twitter_query': 'Búsqueda en Twitter', diff --git a/app/src/lib/i18n/fr.ts b/app/src/lib/i18n/fr.ts index 23cadffe5..0905be256 100644 --- a/app/src/lib/i18n/fr.ts +++ b/app/src/lib/i18n/fr.ts @@ -2082,6 +2082,7 @@ const messages: TranslationMap = { 'memorySources.cssSelectorPlaceholder': 'article', 'memorySources.searchQueryPlaceholder': "de:l'utilisateur AI sécurité", 'memorySources.kind.composio': 'Intégration', + 'memorySources.kind.conversation': 'Conversation', 'memorySources.kind.folder': 'Dossier local', 'memorySources.kind.github_repo': 'GitHub Dépôt', 'memorySources.kind.twitter_query': 'Recherche Twitter', diff --git a/app/src/lib/i18n/hi.ts b/app/src/lib/i18n/hi.ts index 3981fb0c8..b975dad7c 100644 --- a/app/src/lib/i18n/hi.ts +++ b/app/src/lib/i18n/hi.ts @@ -2030,6 +2030,7 @@ const messages: TranslationMap = { 'memorySources.cssSelectorPlaceholder': 'article', 'memorySources.searchQueryPlaceholder': 'से:उपयोगकर्ता AI सुरक्षा', 'memorySources.kind.composio': 'एकीकरण', + 'memorySources.kind.conversation': 'वार्तालाप', 'memorySources.kind.folder': 'स्थानीय फ़ोल्डर', 'memorySources.kind.github_repo': 'GitHub रेपो', 'memorySources.kind.twitter_query': 'Twitter खोज', diff --git a/app/src/lib/i18n/id.ts b/app/src/lib/i18n/id.ts index a093ee460..73597e77e 100644 --- a/app/src/lib/i18n/id.ts +++ b/app/src/lib/i18n/id.ts @@ -2034,6 +2034,7 @@ const messages: TranslationMap = { 'memorySources.cssSelectorPlaceholder': 'article', 'memorySources.searchQueryPlaceholder': 'dari: keselamatan AI pengguna', 'memorySources.kind.composio': 'Integrasi', + 'memorySources.kind.conversation': 'Percakapan', 'memorySources.kind.folder': 'Folder Lokal', 'memorySources.kind.github_repo': 'Repo GitHub', 'memorySources.kind.twitter_query': 'Pencarian Twitter', diff --git a/app/src/lib/i18n/it.ts b/app/src/lib/i18n/it.ts index 08c7085bb..f84cdeec3 100644 --- a/app/src/lib/i18n/it.ts +++ b/app/src/lib/i18n/it.ts @@ -2066,6 +2066,7 @@ const messages: TranslationMap = { 'memorySources.cssSelectorPlaceholder': 'article', 'memorySources.searchQueryPlaceholder': 'da:utente AI sicurezza', 'memorySources.kind.composio': 'Integrazione', + 'memorySources.kind.conversation': 'Conversazione', 'memorySources.kind.folder': 'Cartella Locale', 'memorySources.kind.github_repo': 'GitHub Repository', 'memorySources.kind.twitter_query': 'Ricerca su Twitter', diff --git a/app/src/lib/i18n/ko.ts b/app/src/lib/i18n/ko.ts index 33612f0e6..c27d554e1 100644 --- a/app/src/lib/i18n/ko.ts +++ b/app/src/lib/i18n/ko.ts @@ -2006,6 +2006,7 @@ const messages: TranslationMap = { 'memorySources.cssSelectorPlaceholder': 'article', 'memorySources.searchQueryPlaceholder': 'from:user AI 안전', 'memorySources.kind.composio': '통합', + 'memorySources.kind.conversation': '대화', 'memorySources.kind.folder': '로컬 폴더', 'memorySources.kind.github_repo': 'GitHub 리포지토리', 'memorySources.kind.twitter_query': 'Twitter 검색', diff --git a/app/src/lib/i18n/pl.ts b/app/src/lib/i18n/pl.ts index 882b4d656..cf95f6a54 100644 --- a/app/src/lib/i18n/pl.ts +++ b/app/src/lib/i18n/pl.ts @@ -2051,6 +2051,7 @@ const messages: TranslationMap = { 'memorySources.cssSelectorPlaceholder': 'article', 'memorySources.searchQueryPlaceholder': 'from:user AI safety', 'memorySources.kind.composio': 'Integracja', + 'memorySources.kind.conversation': 'Rozmowa', 'memorySources.kind.folder': 'Folder lokalny', 'memorySources.kind.github_repo': 'Repozytorium GitHub', 'memorySources.kind.twitter_query': 'Wyszukiwanie Twittera', diff --git a/app/src/lib/i18n/pt.ts b/app/src/lib/i18n/pt.ts index 6fa9179e5..d57e304ad 100644 --- a/app/src/lib/i18n/pt.ts +++ b/app/src/lib/i18n/pt.ts @@ -2072,6 +2072,7 @@ const messages: TranslationMap = { 'memorySources.cssSelectorPlaceholder': 'article', 'memorySources.searchQueryPlaceholder': 'de:usuário AI segurança', 'memorySources.kind.composio': 'Integração', + 'memorySources.kind.conversation': 'Conversa', 'memorySources.kind.folder': 'Pasta Local', 'memorySources.kind.github_repo': 'Repositório GitHub', 'memorySources.kind.twitter_query': 'Busca no Twitter', diff --git a/app/src/lib/i18n/ru.ts b/app/src/lib/i18n/ru.ts index c0353068f..02f8c6c83 100644 --- a/app/src/lib/i18n/ru.ts +++ b/app/src/lib/i18n/ru.ts @@ -2044,6 +2044,7 @@ const messages: TranslationMap = { 'memorySources.cssSelectorPlaceholder': 'article', 'memorySources.searchQueryPlaceholder': 'от:пользователь AI безопасность', 'memorySources.kind.composio': 'Интеграция', + 'memorySources.kind.conversation': 'Разговор', 'memorySources.kind.folder': 'Локальная папка', 'memorySources.kind.github_repo': 'Репо GitHub', 'memorySources.kind.twitter_query': 'Поиск в Твиттере', diff --git a/app/src/lib/i18n/zh-CN.ts b/app/src/lib/i18n/zh-CN.ts index 588cd4572..4f3761a58 100644 --- a/app/src/lib/i18n/zh-CN.ts +++ b/app/src/lib/i18n/zh-CN.ts @@ -1924,6 +1924,7 @@ const messages: TranslationMap = { 'memorySources.cssSelectorPlaceholder': 'article', 'memorySources.searchQueryPlaceholder': 'from:user AI safety', 'memorySources.kind.composio': '集成', + 'memorySources.kind.conversation': '对话', 'memorySources.kind.folder': '本地文件夹', 'memorySources.kind.github_repo': 'GitHub 仓库', 'memorySources.kind.twitter_query': 'Twitter 搜索', diff --git a/app/src/services/memorySourcesService.test.ts b/app/src/services/memorySourcesService.test.ts index cf2a499a1..ff4519644 100644 --- a/app/src/services/memorySourcesService.test.ts +++ b/app/src/services/memorySourcesService.test.ts @@ -87,6 +87,7 @@ describe('memorySourcesService', () => { it('exposes labels and icons for every source kind', () => { const kinds = [ 'composio', + 'conversation', 'folder', 'github_repo', 'twitter_query', @@ -99,6 +100,28 @@ describe('memorySourcesService', () => { } }); + it('addMemorySource sends conversation source with no extra fields', async () => { + mockedCall.mockResolvedValue({ + result: { + source: { id: 'src_conv', kind: 'conversation', label: 'Conversations', enabled: true }, + }, + logs: [], + } as never); + + const result = await addMemorySource({ + kind: 'conversation', + label: 'Conversations', + enabled: true, + }); + + expect(mockedCall).toHaveBeenCalledWith({ + method: 'openhuman.memory_sources_add', + params: { kind: 'conversation', label: 'Conversations', enabled: true }, + }); + expect(result.id).toBe('src_conv'); + expect(result.kind).toBe('conversation'); + }); + // --------------------------------------------------------------------------- // applyAllIn // --------------------------------------------------------------------------- diff --git a/app/src/services/memorySourcesService.ts b/app/src/services/memorySourcesService.ts index 47f6023b9..bcd646a5b 100644 --- a/app/src/services/memorySourcesService.ts +++ b/app/src/services/memorySourcesService.ts @@ -12,6 +12,7 @@ const log = debug('memory-sources'); export type SourceKind = | 'composio' + | 'conversation' | 'folder' | 'github_repo' | 'twitter_query' @@ -206,6 +207,7 @@ export async function applyAllIn(): Promise { /// without each call site duplicating the switch. export const SOURCE_KIND_LABEL_KEYS: Record = { composio: 'memorySources.kind.composio', + conversation: 'memorySources.kind.conversation', folder: 'memorySources.kind.folder', github_repo: 'memorySources.kind.github_repo', twitter_query: 'memorySources.kind.twitter_query', @@ -215,6 +217,7 @@ export const SOURCE_KIND_LABEL_KEYS: Record = { export const SOURCE_KIND_ICONS: Record = { composio: '🔗', + conversation: '💬', folder: '📁', github_repo: '🐙', twitter_query: '🐦', diff --git a/app/test/e2e/specs/memory-sources-conversation.spec.ts b/app/test/e2e/specs/memory-sources-conversation.spec.ts new file mode 100644 index 000000000..ff66656eb --- /dev/null +++ b/app/test/e2e/specs/memory-sources-conversation.spec.ts @@ -0,0 +1,110 @@ +/** + * Memory sources — Conversation kind CRUD via RPC. + * + * Verifies the conversation source kind can be added, listed, toggled, + * and removed via the core JSON-RPC surface. Does not test the full UI + * wizard (covered by unit tests) — focuses on the RPC round-trip that + * backs it. + * + * Flow: + * 1. Add a conversation source via RPC + * 2. List sources and verify it appears + * 3. Get source by id + * 4. Update (disable) the source + * 5. Remove the source + * 6. List confirms removal + */ +import { waitForApp } from '../helpers/app-helpers'; +import { callOpenhumanRpc, expectRpcOk } from '../helpers/core-rpc'; +import { resetApp } from '../helpers/reset-app'; + +const LOG_PREFIX = '[memory-sources-conversation]'; +const USER_ID = 'e2e-memory-sources-conversation'; + +interface MemorySource { + id: string; + kind: string; + label: string; + enabled: boolean; +} + +describe('Memory sources — conversation kind', () => { + let sourceId: string; + + before(async () => { + console.log(`${LOG_PREFIX} Waiting for app`); + await waitForApp(); + await resetApp(USER_ID); + console.log(`${LOG_PREFIX} Setup complete`); + }); + + it('adds a conversation source via RPC', async () => { + const resp = await callOpenhumanRpc<{ source: MemorySource }>('openhuman.memory_sources_add', { + kind: 'conversation', + label: 'Agent Conversations', + enabled: true, + }); + expectRpcOk('openhuman.memory_sources_add', resp); + const data = resp.result!; + expect(data.source).toBeDefined(); + expect(data.source.kind).toBe('conversation'); + expect(data.source.label).toBe('Agent Conversations'); + expect(data.source.enabled).toBe(true); + expect(data.source.id).toBeTruthy(); + sourceId = data.source.id; + console.log(`${LOG_PREFIX} Added source: ${sourceId}`); + }); + + it('lists sources including the conversation source', async () => { + const resp = await callOpenhumanRpc<{ sources: MemorySource[] }>( + 'openhuman.memory_sources_list', + {} + ); + expectRpcOk('openhuman.memory_sources_list', resp); + const sources = resp.result!.sources ?? []; + const convSources = sources.filter(s => s.kind === 'conversation'); + expect(convSources.length).toBeGreaterThanOrEqual(1); + expect(convSources[0].id).toBe(sourceId); + }); + + it('gets the source by id', async () => { + const resp = await callOpenhumanRpc<{ source: MemorySource }>('openhuman.memory_sources_get', { + id: sourceId, + }); + expectRpcOk('openhuman.memory_sources_get', resp); + const data = resp.result!; + expect(data.source).toBeDefined(); + expect(data.source.kind).toBe('conversation'); + expect(data.source.id).toBe(sourceId); + }); + + it('updates the source to disabled', async () => { + const resp = await callOpenhumanRpc<{ source: MemorySource }>( + 'openhuman.memory_sources_update', + { id: sourceId, enabled: false } + ); + expectRpcOk('openhuman.memory_sources_update', resp); + const data = resp.result!; + expect(data.source.enabled).toBe(false); + expect(data.source.kind).toBe('conversation'); + }); + + it('removes the conversation source', async () => { + const resp = await callOpenhumanRpc<{ removed: boolean }>('openhuman.memory_sources_remove', { + id: sourceId, + }); + expectRpcOk('openhuman.memory_sources_remove', resp); + expect(resp.result!.removed).toBe(true); + }); + + it('list confirms source is removed', async () => { + const resp = await callOpenhumanRpc<{ sources: MemorySource[] }>( + 'openhuman.memory_sources_list', + {} + ); + expectRpcOk('openhuman.memory_sources_list', resp); + const sources = resp.result!.sources ?? []; + const convSources = sources.filter(s => s.kind === 'conversation'); + expect(convSources.length).toBe(0); + }); +}); diff --git a/src/openhuman/memory_sources/readers/conversation.rs b/src/openhuman/memory_sources/readers/conversation.rs new file mode 100644 index 000000000..1d8ea92f8 --- /dev/null +++ b/src/openhuman/memory_sources/readers/conversation.rs @@ -0,0 +1,378 @@ +//! Conversation source reader. +//! +//! Treats every agent conversation as a memory source item. When synced, +//! each conversation's messages are stored as durable memory alongside +//! other sources like GitHub, integrations, etc. + +use async_trait::async_trait; + +use crate::openhuman::config::Config; +use crate::openhuman::memory_sources::types::{ + ContentType, MemorySourceEntry, SourceContent, SourceItem, SourceKind, +}; + +use super::SourceReader; + +pub struct ConversationReader; + +#[async_trait] +impl SourceReader for ConversationReader { + fn kind(&self) -> SourceKind { + SourceKind::Conversation + } + + async fn list_items( + &self, + _source: &MemorySourceEntry, + config: &Config, + ) -> Result, String> { + tracing::debug!("[memory_sources:conversation] list_items"); + + let threads_dir = config.workspace_dir.join("threads"); + if !threads_dir.exists() { + return Ok(Vec::new()); + } + + let mut items = Vec::new(); + let mut entries = tokio::fs::read_dir(&threads_dir) + .await + .map_err(|e| format!("failed to read threads dir: {e}"))?; + + loop { + match entries.next_entry().await { + Ok(Some(entry)) => { + let path = entry.path(); + if path.extension().and_then(|e| e.to_str()) != Some("json") { + continue; + } + let id = path + .file_stem() + .and_then(|s| s.to_str()) + .unwrap_or_default() + .to_string(); + + let modified_ms = entry + .metadata() + .await + .ok() + .and_then(|m| m.modified().ok()) + .and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok()) + .map(|d| d.as_millis() as i64); + + items.push(SourceItem { + id, + title: path + .file_stem() + .and_then(|s| s.to_str()) + .unwrap_or("conversation") + .to_string(), + updated_at_ms: modified_ms, + }); + } + Ok(None) => break, + Err(err) => { + tracing::warn!( + error = %err, + "[memory_sources:conversation] failed to read directory entry, skipping" + ); + continue; + } + } + } + + tracing::debug!( + count = items.len(), + "[memory_sources:conversation] found threads" + ); + + Ok(items) + } + + async fn read_item( + &self, + _source: &MemorySourceEntry, + item_id: &str, + config: &Config, + ) -> Result { + tracing::debug!( + item_id = %item_id, + "[memory_sources:conversation] read_item" + ); + + // Validate item_id to prevent path traversal + if item_id.contains("..") || item_id.contains('/') || item_id.contains('\\') { + return Err("invalid item_id: path traversal denied".to_string()); + } + + let threads_dir = config.workspace_dir.join("threads"); + let thread_path = threads_dir.join(format!("{item_id}.json")); + + // Canonicalize and verify containment within threads directory + if !thread_path.exists() { + return Err(format!("thread '{item_id}' not found")); + } + let canonical_base = std::fs::canonicalize(&threads_dir) + .map_err(|e| format!("cannot resolve threads dir: {e}"))?; + let canonical_file = std::fs::canonicalize(&thread_path) + .map_err(|e| format!("cannot resolve thread path: {e}"))?; + if !canonical_file.starts_with(&canonical_base) { + return Err("path traversal denied".to_string()); + } + + let raw = tokio::fs::read_to_string(&canonical_file) + .await + .map_err(|e| format!("failed to read thread file: {e}"))?; + + let parsed: serde_json::Value = + serde_json::from_str(&raw).map_err(|e| format!("failed to parse thread JSON: {e}"))?; + + let title = parsed + .get("title") + .and_then(|v| v.as_str()) + .unwrap_or(item_id) + .to_string(); + + let body = format_thread_as_markdown(&parsed); + + Ok(SourceContent { + id: item_id.to_string(), + title, + body, + content_type: ContentType::Markdown, + metadata: serde_json::json!({ + "source_type": "conversation", + "thread_id": item_id, + }), + }) + } +} + +fn format_thread_as_markdown(thread: &serde_json::Value) -> String { + let mut out = String::new(); + + if let Some(title) = thread.get("title").and_then(|v| v.as_str()) { + out.push_str(&format!("# {title}\n\n")); + } + + let messages = thread + .get("messages") + .and_then(|v| v.as_array()) + .cloned() + .unwrap_or_default(); + + for msg in &messages { + let role = msg + .get("role") + .and_then(|v| v.as_str()) + .unwrap_or("unknown"); + let content = msg.get("content").and_then(|v| v.as_str()).unwrap_or(""); + + if content.is_empty() { + continue; + } + + out.push_str(&format!("**{role}**: {content}\n\n")); + } + + out +} + +#[cfg(test)] +mod tests { + use super::*; + use std::fs; + use tempfile::tempdir; + + #[test] + fn format_thread_produces_markdown() { + let thread = serde_json::json!({ + "title": "Test chat", + "messages": [ + {"role": "user", "content": "Hello"}, + {"role": "assistant", "content": "Hi there!"}, + ] + }); + let md = format_thread_as_markdown(&thread); + assert!(md.contains("# Test chat")); + assert!(md.contains("**user**: Hello")); + assert!(md.contains("**assistant**: Hi there!")); + } + + #[test] + fn format_thread_skips_empty_content() { + let thread = serde_json::json!({ + "title": "Sparse", + "messages": [ + {"role": "user", "content": ""}, + {"role": "assistant", "content": "Reply"}, + {"role": "user", "content": ""}, + ] + }); + let md = format_thread_as_markdown(&thread); + assert!(!md.contains("**user**:")); + assert!(md.contains("**assistant**: Reply")); + } + + #[test] + fn format_thread_handles_missing_title() { + let thread = serde_json::json!({ + "messages": [{"role": "user", "content": "Hi"}] + }); + let md = format_thread_as_markdown(&thread); + assert!(!md.starts_with('#')); + assert!(md.contains("**user**: Hi")); + } + + #[test] + fn format_thread_handles_no_messages() { + let thread = serde_json::json!({"title": "Empty"}); + let md = format_thread_as_markdown(&thread); + assert!(md.contains("# Empty")); + assert_eq!(md.trim(), "# Empty"); + } + + #[tokio::test] + async fn list_items_returns_empty_when_no_threads_dir() { + let tmp = tempdir().unwrap(); + let mut config = Config::default(); + config.workspace_dir = tmp.path().to_path_buf(); + + let source = conversation_source(); + let reader = ConversationReader; + let items = reader.list_items(&source, &config).await.unwrap(); + assert!(items.is_empty()); + } + + #[tokio::test] + async fn list_items_finds_json_thread_files() { + let tmp = tempdir().unwrap(); + let threads_dir = tmp.path().join("threads"); + fs::create_dir_all(&threads_dir).unwrap(); + + fs::write( + threads_dir.join("thread_abc.json"), + r#"{"title":"Chat 1","messages":[]}"#, + ) + .unwrap(); + fs::write( + threads_dir.join("thread_def.json"), + r#"{"title":"Chat 2","messages":[]}"#, + ) + .unwrap(); + // Non-json file should be ignored + fs::write(threads_dir.join("notes.txt"), "ignored").unwrap(); + + let mut config = Config::default(); + config.workspace_dir = tmp.path().to_path_buf(); + + let source = conversation_source(); + let reader = ConversationReader; + let items = reader.list_items(&source, &config).await.unwrap(); + assert_eq!(items.len(), 2); + + let ids: Vec<&str> = items.iter().map(|i| i.id.as_str()).collect(); + assert!(ids.contains(&"thread_abc")); + assert!(ids.contains(&"thread_def")); + } + + #[tokio::test] + async fn read_item_returns_formatted_content() { + let tmp = tempdir().unwrap(); + let threads_dir = tmp.path().join("threads"); + fs::create_dir_all(&threads_dir).unwrap(); + + let thread_json = serde_json::json!({ + "title": "Test Conversation", + "messages": [ + {"role": "user", "content": "What is 2+2?"}, + {"role": "assistant", "content": "4"}, + ] + }); + fs::write( + threads_dir.join("conv_123.json"), + serde_json::to_string(&thread_json).unwrap(), + ) + .unwrap(); + + let mut config = Config::default(); + config.workspace_dir = tmp.path().to_path_buf(); + + let source = conversation_source(); + let reader = ConversationReader; + let content = reader + .read_item(&source, "conv_123", &config) + .await + .unwrap(); + + assert_eq!(content.id, "conv_123"); + assert_eq!(content.title, "Test Conversation"); + assert_eq!(content.content_type, ContentType::Markdown); + assert!(content.body.contains("**user**: What is 2+2?")); + assert!(content.body.contains("**assistant**: 4")); + } + + #[tokio::test] + async fn read_item_returns_error_for_missing_thread() { + let tmp = tempdir().unwrap(); + let threads_dir = tmp.path().join("threads"); + fs::create_dir_all(&threads_dir).unwrap(); + + let mut config = Config::default(); + config.workspace_dir = tmp.path().to_path_buf(); + + let source = conversation_source(); + let reader = ConversationReader; + let result = reader.read_item(&source, "nonexistent", &config).await; + assert!(result.is_err()); + assert!(result.unwrap_err().contains("not found")); + } + + #[tokio::test] + async fn read_item_rejects_path_traversal() { + let tmp = tempdir().unwrap(); + let threads_dir = tmp.path().join("threads"); + fs::create_dir_all(&threads_dir).unwrap(); + + let mut config = Config::default(); + config.workspace_dir = tmp.path().to_path_buf(); + + let source = conversation_source(); + let reader = ConversationReader; + + let result = reader.read_item(&source, "../config", &config).await; + assert!(result.is_err()); + assert!(result.unwrap_err().contains("path traversal denied")); + + let result = reader + .read_item(&source, "foo/../../etc/passwd", &config) + .await; + assert!(result.is_err()); + assert!(result.unwrap_err().contains("path traversal denied")); + } + + fn conversation_source() -> MemorySourceEntry { + MemorySourceEntry { + id: "src_conv".into(), + kind: SourceKind::Conversation, + label: "Conversations".into(), + enabled: true, + toolkit: None, + connection_id: None, + path: None, + glob: None, + url: None, + branch: None, + paths: Vec::new(), + query: None, + since_days: None, + max_items: None, + max_commits: None, + max_issues: None, + max_prs: None, + selector: None, + max_tokens_per_sync: None, + max_cost_per_sync_usd: None, + sync_depth_days: None, + } + } +} diff --git a/src/openhuman/memory_sources/readers/mod.rs b/src/openhuman/memory_sources/readers/mod.rs index 1015bd2bb..e7373dbb5 100644 --- a/src/openhuman/memory_sources/readers/mod.rs +++ b/src/openhuman/memory_sources/readers/mod.rs @@ -1,6 +1,7 @@ //! Source reader trait and per-kind implementations. pub mod composio; +pub mod conversation; pub mod folder; pub mod github; pub mod rss; @@ -35,6 +36,7 @@ pub trait SourceReader: Send + Sync { pub fn reader_for(kind: &SourceKind) -> Box { match kind { SourceKind::Composio => Box::new(composio::ComposioReader), + SourceKind::Conversation => Box::new(conversation::ConversationReader), SourceKind::Folder => Box::new(folder::FolderReader), SourceKind::GithubRepo => Box::new(github::GithubReader), SourceKind::TwitterQuery => Box::new(twitter::TwitterReader), diff --git a/src/openhuman/memory_sources/schemas.rs b/src/openhuman/memory_sources/schemas.rs index a35950e69..6b218aba3 100644 --- a/src/openhuman/memory_sources/schemas.rs +++ b/src/openhuman/memory_sources/schemas.rs @@ -236,6 +236,7 @@ pub fn schemas(function: &str) -> ControllerSchema { ty: TypeSchema::Enum { variants: vec![ "composio", + "conversation", "folder", "github_repo", "twitter_query", diff --git a/src/openhuman/memory_sources/sync.rs b/src/openhuman/memory_sources/sync.rs index 49bede949..22899b047 100644 --- a/src/openhuman/memory_sources/sync.rs +++ b/src/openhuman/memory_sources/sync.rs @@ -97,6 +97,7 @@ pub async fn sync_source(source: MemorySourceEntry, config: Config) -> Result<() SourceKind::Composio => { sync_composio(&source, config.clone(), &mut composio_usage).await } + SourceKind::Conversation => sync_items_individually(&source, &config).await, SourceKind::GithubRepo => { // GitHub path writes its own detailed audit entry // with token breakdowns; skip the dispatcher-level diff --git a/src/openhuman/memory_sources/types.rs b/src/openhuman/memory_sources/types.rs index f0bce598a..258472aa0 100644 --- a/src/openhuman/memory_sources/types.rs +++ b/src/openhuman/memory_sources/types.rs @@ -11,6 +11,7 @@ fn default_true() -> bool { #[serde(rename_all = "snake_case")] pub enum SourceKind { Composio, + Conversation, Folder, GithubRepo, TwitterQuery, @@ -22,6 +23,7 @@ impl SourceKind { pub fn as_str(&self) -> &'static str { match self { SourceKind::Composio => "composio", + SourceKind::Conversation => "conversation", SourceKind::Folder => "folder", SourceKind::GithubRepo => "github_repo", SourceKind::TwitterQuery => "twitter_query", @@ -114,6 +116,9 @@ impl MemorySourceEntry { require_field(&self.toolkit, "toolkit")?; require_field(&self.connection_id, "connection_id")?; } + SourceKind::Conversation => { + // No kind-specific required fields — just enabled/disabled. + } SourceKind::Folder => { require_field(&self.path, "path")?; } @@ -177,6 +182,7 @@ mod tests { fn source_kind_round_trips_via_serde() { for kind in [ SourceKind::Composio, + SourceKind::Conversation, SourceKind::Folder, SourceKind::GithubRepo, SourceKind::TwitterQuery, @@ -235,6 +241,48 @@ mod tests { assert!(entry.validate().is_ok()); } + #[test] + fn validate_conversation_needs_only_id_and_label() { + let entry = MemorySourceEntry { + id: "src_conv".into(), + kind: SourceKind::Conversation, + label: "Agent Conversations".into(), + enabled: true, + ..default_entry() + }; + assert!(entry.validate().is_ok()); + } + + #[test] + fn validate_conversation_fails_with_empty_id() { + let entry = MemorySourceEntry { + id: "".into(), + kind: SourceKind::Conversation, + label: "Convos".into(), + enabled: true, + ..default_entry() + }; + assert!(entry.validate().is_err()); + } + + #[test] + fn validate_conversation_fails_with_empty_label() { + let entry = MemorySourceEntry { + id: "src_conv".into(), + kind: SourceKind::Conversation, + label: "".into(), + enabled: true, + ..default_entry() + }; + assert!(entry.validate().is_err()); + } + + #[test] + fn conversation_kind_serializes_to_snake_case() { + let json = serde_json::to_string(&SourceKind::Conversation).unwrap(); + assert_eq!(json, "\"conversation\""); + } + #[test] fn toml_round_trip() { let entry = MemorySourceEntry { @@ -253,6 +301,23 @@ mod tests { assert_eq!(decoded.path.as_deref(), Some("/tmp/notes")); } + #[test] + fn conversation_toml_round_trip() { + let entry = MemorySourceEntry { + id: "src_conv".into(), + kind: SourceKind::Conversation, + label: "Conversations".into(), + enabled: true, + ..default_entry() + }; + let toml_str = toml::to_string_pretty(&entry).unwrap(); + let decoded: MemorySourceEntry = toml::from_str(&toml_str).unwrap(); + assert_eq!(decoded.id, "src_conv"); + assert_eq!(decoded.kind, SourceKind::Conversation); + assert_eq!(decoded.label, "Conversations"); + assert!(decoded.enabled); + } + fn default_entry() -> MemorySourceEntry { MemorySourceEntry { id: String::new(), diff --git a/src/openhuman/memory_store/chunks/types.rs b/src/openhuman/memory_store/chunks/types.rs index 7ee1ececa..f96e8f872 100644 --- a/src/openhuman/memory_store/chunks/types.rs +++ b/src/openhuman/memory_store/chunks/types.rs @@ -68,6 +68,9 @@ pub enum DataSource { Telegram, Whatsapp, + // ── Agent conversations (stored as durable memory) ──────────────── + Conversation, + // ── Email threads (grouped by thread) ────────────────────────────── Gmail, /// Catch-all for non-Gmail providers (Outlook, FastMail, generic IMAP, …). @@ -83,7 +86,9 @@ impl DataSource { /// Which [`SourceKind`] this provider feeds into. pub fn kind(self) -> SourceKind { match self { - Self::Discord | Self::Telegram | Self::Whatsapp => SourceKind::Chat, + Self::Discord | Self::Telegram | Self::Whatsapp | Self::Conversation => { + SourceKind::Chat + } Self::Gmail | Self::OtherEmail => SourceKind::Email, Self::Notion | Self::MeetingNotes | Self::DriveDocs => SourceKind::Document, } @@ -95,6 +100,7 @@ impl DataSource { Self::Discord => "discord", Self::Telegram => "telegram", Self::Whatsapp => "whatsapp", + Self::Conversation => "conversation", Self::Gmail => "gmail", Self::OtherEmail => "other_email", Self::Notion => "notion", @@ -109,6 +115,7 @@ impl DataSource { "discord" => Ok(Self::Discord), "telegram" => Ok(Self::Telegram), "whatsapp" => Ok(Self::Whatsapp), + "conversation" => Ok(Self::Conversation), "gmail" => Ok(Self::Gmail), "other_email" => Ok(Self::OtherEmail), "notion" => Ok(Self::Notion), @@ -127,6 +134,7 @@ impl DataSource { Self::Discord, Self::Telegram, Self::Whatsapp, + Self::Conversation, Self::Gmail, Self::OtherEmail, Self::Notion, @@ -386,15 +394,14 @@ mod tests { } #[test] - fn data_source_has_all_eight_variants_from_m_excalidraw() { - // Guard against accidental drift from the canonical provider list. - assert_eq!(DataSource::all().len(), 8); + fn data_source_has_all_variants() { + assert_eq!(DataSource::all().len(), 9); } #[test] fn data_source_kind_mapping() { use DataSource::*; - for ds in [Discord, Telegram, Whatsapp] { + for ds in [Discord, Telegram, Whatsapp, Conversation] { assert_eq!(ds.kind(), SourceKind::Chat); } for ds in [Gmail, OtherEmail] { diff --git a/src/openhuman/memory_tree/score/signals/source_weight.rs b/src/openhuman/memory_tree/score/signals/source_weight.rs index b9180fa67..80cd30770 100644 --- a/src/openhuman/memory_tree/score/signals/source_weight.rs +++ b/src/openhuman/memory_tree/score/signals/source_weight.rs @@ -56,6 +56,8 @@ fn weight_for(ds: DataSource) -> f32 { DataSource::Whatsapp => 0.75, DataSource::Telegram => 0.6, DataSource::Discord => 0.5, + // Agent conversations — high signal, direct interaction with the user + DataSource::Conversation => 0.9, // Documents: Notion = structured, Drive = mixed, Meeting notes = high value DataSource::Notion => 0.75, DataSource::DriveDocs => 0.6, diff --git a/tests/json_rpc_e2e.rs b/tests/json_rpc_e2e.rs index c47eaf596..b35dca771 100644 --- a/tests/json_rpc_e2e.rs +++ b/tests/json_rpc_e2e.rs @@ -10885,3 +10885,133 @@ async fn json_rpc_memory_sync_settings_env_override_is_reflected() { rpc_join.abort(); } + +#[tokio::test] +async fn json_rpc_memory_sources_conversation_crud() { + let _env_lock = json_rpc_e2e_env_lock(); + let tmp = tempdir().expect("tempdir"); + let home = tmp.path(); + let _home_guard = EnvVarGuard::set_to_path("HOME", home); + let _ws_guard = EnvVarGuard::set_to_path("OPENHUMAN_WORKSPACE", home); + let _action_guard = EnvVarGuard::set_to_path("OPENHUMAN_ACTION_DIR", home); + let _backend_guard = EnvVarGuard::unset("VITE_BACKEND_URL"); + + let (rpc_addr, rpc_join) = serve_on_ephemeral(build_core_http_router(false)).await; + let rpc_base = format!("http://{rpc_addr}"); + + // 1. Add a conversation source + let add_resp = post_json_rpc( + &rpc_base, + 9501, + "openhuman.memory_sources_add", + json!({ + "kind": "conversation", + "label": "Agent Conversations", + "enabled": true + }), + ) + .await; + let add_result = assert_no_jsonrpc_error(&add_resp, "memory_sources_add conversation"); + let source = add_result + .get("source") + .expect("add should return source object"); + let source_id = source + .get("id") + .and_then(Value::as_str) + .expect("source must have id"); + assert_eq!( + source.get("kind").and_then(Value::as_str), + Some("conversation") + ); + assert_eq!( + source.get("label").and_then(Value::as_str), + Some("Agent Conversations") + ); + assert_eq!(source.get("enabled").and_then(Value::as_bool), Some(true)); + + // 2. List sources — should contain our conversation source + let list_resp = + post_json_rpc(&rpc_base, 9502, "openhuman.memory_sources_list", json!({})).await; + let list_result = assert_no_jsonrpc_error(&list_resp, "memory_sources_list"); + let sources = list_result + .get("sources") + .and_then(Value::as_array) + .expect("list should return sources array"); + let conversation_sources: Vec<&Value> = sources + .iter() + .filter(|s| s.get("kind").and_then(Value::as_str) == Some("conversation")) + .collect(); + assert_eq!( + conversation_sources.len(), + 1, + "exactly one conversation source after add" + ); + + // 3. Get the source by id + let get_resp = post_json_rpc( + &rpc_base, + 9503, + "openhuman.memory_sources_get", + json!({ "id": source_id }), + ) + .await; + let get_result = assert_no_jsonrpc_error(&get_resp, "memory_sources_get"); + assert_eq!( + get_result + .get("source") + .and_then(|s| s.get("kind")) + .and_then(Value::as_str), + Some("conversation") + ); + + // 4. Update — disable the source + let update_resp = post_json_rpc( + &rpc_base, + 9504, + "openhuman.memory_sources_update", + json!({ "id": source_id, "enabled": false }), + ) + .await; + let update_result = assert_no_jsonrpc_error(&update_resp, "memory_sources_update"); + assert_eq!( + update_result + .get("source") + .and_then(|s| s.get("enabled")) + .and_then(Value::as_bool), + Some(false), + "source should be disabled after update" + ); + + // 5. Remove the source + let remove_resp = post_json_rpc( + &rpc_base, + 9505, + "openhuman.memory_sources_remove", + json!({ "id": source_id }), + ) + .await; + let remove_result = assert_no_jsonrpc_error(&remove_resp, "memory_sources_remove"); + assert_eq!( + remove_result.get("removed").and_then(Value::as_bool), + Some(true) + ); + + // 6. List again — empty + let list2_resp = + post_json_rpc(&rpc_base, 9506, "openhuman.memory_sources_list", json!({})).await; + let list2_result = assert_no_jsonrpc_error(&list2_resp, "memory_sources_list after remove"); + let sources2 = list2_result + .get("sources") + .and_then(Value::as_array) + .expect("list should return sources array"); + let conv_remaining: Vec<&Value> = sources2 + .iter() + .filter(|s| s.get("kind").and_then(Value::as_str) == Some("conversation")) + .collect(); + assert!( + conv_remaining.is_empty(), + "no conversation sources after remove" + ); + + rpc_join.abort(); +}