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

feat(memory): add Conversation as a memory source kind (#3441)

Browse Source
This commit is contained in:
Steven Enamakel's Droid
2026-06-06 15:55:20 -04:00
committed by GitHub
parent e445aa4c50
commit 6cf53c0282
28 changed files with 792 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/src/components/intelligence/AddMemorySourceDialog.test.tsx
+43
View File
@@ -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<typeof vi.fn>;
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
// ---------------------------------------------------------------------------
app/src/components/intelligence/AddMemorySourceDialog.tsx
+7
View File
@@ -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 <ComposioPicker {...props} />;
case 'conversation':
return null;
case 'folder':
return (
<>
app/src/components/intelligence/SourceSettingsPanel.tsx
+1
View File
@@ -24,6 +24,7 @@ import {
// are intentionally omitted here.
const KIND_FIELDS: Record<SourceKind, Array<keyof LimitFields>> = {
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'],
app/src/lib/i18n/ar.ts
+1
View File
@@ -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',
app/src/lib/i18n/bn.ts
+1
View File
@@ -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 অনুসন্ধান',
app/src/lib/i18n/de.ts
+1
View File
@@ -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',
app/src/lib/i18n/en.ts
+1
View File
@@ -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',
app/src/lib/i18n/es.ts
+1
View File
@@ -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',
app/src/lib/i18n/fr.ts
+1
View File
@@ -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',
app/src/lib/i18n/hi.ts
+1
View File
@@ -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 खोज',
app/src/lib/i18n/id.ts
+1
View File
@@ -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',
app/src/lib/i18n/it.ts
+1
View File
@@ -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',
app/src/lib/i18n/ko.ts
+1
View File
@@ -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 검색',
app/src/lib/i18n/pl.ts
+1
View File
@@ -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',
app/src/lib/i18n/pt.ts
+1
View File
@@ -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',
app/src/lib/i18n/ru.ts
+1
View File
@@ -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': 'Поиск в Твиттере',
app/src/lib/i18n/zh-CN.ts
+1
View File
@@ -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 搜索',
app/src/services/memorySourcesService.test.ts
+23
View File
@@ -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
// ---------------------------------------------------------------------------
app/src/services/memorySourcesService.ts
+3
View File
@@ -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<ApplyAllInResult> {
/// without each call site duplicating the switch.
export const SOURCE_KIND_LABEL_KEYS: Record<SourceKind, string> = {
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<SourceKind, string> = {
export const SOURCE_KIND_ICONS: Record<SourceKind, string> = {
composio: '🔗',
conversation: '💬',
folder: '📁',
github_repo: '🐙',
twitter_query: '🐦',
app/test/e2e/specs/memory-sources-conversation.spec.ts
+110
View File
@@ -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);
});
});
src/openhuman/memory_sources/readers/conversation.rs
+378
View File
@@ -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<Vec<SourceItem>, 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<SourceContent, String> {
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,
}
}
}
src/openhuman/memory_sources/readers/mod.rs
+2
View File
@@ -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<dyn SourceReader> {
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),
src/openhuman/memory_sources/schemas.rs
+1
View File
@@ -236,6 +236,7 @@ pub fn schemas(function: &str) -> ControllerSchema {
ty: TypeSchema::Enum {
variants: vec![
"composio",
"conversation",
"folder",
"github_repo",
"twitter_query",
src/openhuman/memory_sources/sync.rs
+1
View File
@@ -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
src/openhuman/memory_sources/types.rs
+65
View File
@@ -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(),
src/openhuman/memory_store/chunks/types.rs
+12 -5
View File
@@ -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] {
src/openhuman/memory_tree/score/signals/source_weight.rs
+2
View File
@@ -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,
tests/json_rpc_e2e.rs
+130
View File
@@ -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();
}