From 2304d9d58fee9d734c2d8c50240ddc625f349449 Mon Sep 17 00:00:00 2001 From: Steven Enamakel <31011319+senamakel@users.noreply.github.com> Date: Wed, 29 Apr 2026 18:43:11 -0700 Subject: [PATCH] feat(memory): add Sync and Learn sections to Memory workspace (#1031) --- .../intelligence/MemoryWorkspace.tsx | 262 ++++++++++++++++++ .../__tests__/MemoryWorkspace.test.tsx | 111 +++++++- app/src/utils/tauriCommands/memory.test.ts | 111 ++++++++ app/src/utils/tauriCommands/memory.ts | 73 +++++ src/core/event_bus/events.rs | 11 +- src/openhuman/config/schema/load_tests.rs | 44 ++- src/openhuman/memory/ops.rs | 210 ++++++++++++++ src/openhuman/memory/schemas.rs | 101 ++++++- src/openhuman/memory/schemas_tests.rs | 39 +++ tests/json_rpc_e2e.rs | 99 +++++++ 10 files changed, 1047 insertions(+), 14 deletions(-) create mode 100644 app/src/utils/tauriCommands/memory.test.ts diff --git a/app/src/components/intelligence/MemoryWorkspace.tsx b/app/src/components/intelligence/MemoryWorkspace.tsx index 6ac95eb3d..cb89d4907 100644 --- a/app/src/components/intelligence/MemoryWorkspace.tsx +++ b/app/src/components/intelligence/MemoryWorkspace.tsx @@ -1,6 +1,8 @@ import { useCallback, useEffect, useMemo, useState } from 'react'; import { useIntelligenceStats } from '../../hooks/useIntelligenceStats'; +import { channelConnectionsApi } from '../../services/api/channelConnectionsApi'; +import type { ChannelStatusEntry } from '../../types/channels'; import type { ToastNotification } from '../../types/intelligence'; import { aiListMemoryFiles, @@ -10,11 +12,15 @@ import { isTauri, memoryDeleteDocument, memoryGraphQuery, + memoryLearnAll, + type MemoryLearnAllResult, memoryListDocuments, memoryListNamespaces, memoryQueryNamespace, type MemoryQueryResult, memoryRecallNamespace, + memorySyncAll, + memorySyncChannel, } from '../../utils/tauriCommands'; import { MemoryGraphMap } from './MemoryGraphMap'; import { MemoryHeatmap } from './MemoryHeatmap'; @@ -132,6 +138,18 @@ export function MemoryWorkspace({ onToast }: MemoryWorkspaceProps) { const [graphRelations, setGraphRelations] = useState([]); const [graphRelationsLoading, setGraphRelationsLoading] = useState(false); + // Sync section + const [syncOpen, setSyncOpen] = useState(false); + const [connectedChannels, setConnectedChannels] = useState([]); + const [syncingAll, setSyncingAll] = useState(false); + const [syncingChannelId, setSyncingChannelId] = useState(null); + + // Learn section + const [learnOpen, setLearnOpen] = useState(false); + const [learning, setLearning] = useState(false); + const [learnResult, setLearnResult] = useState(null); + const [learnErrorOpen, setLearnErrorOpen] = useState(false); + // Manage memory section (collapsible) const [manageOpen, setManageOpen] = useState(false); const [selectedNamespace, setSelectedNamespace] = useState(''); @@ -180,6 +198,15 @@ export function MemoryWorkspace({ onToast }: MemoryWorkspaceProps) { const docs = normalizeMemoryDocuments(documentsPayload); const combinedFiles = ['memory.md', ...memoryDirFiles.map(file => `memory/${file}`)]; + // Load connected channels for the Sync section. + try { + const statuses = await channelConnectionsApi.listStatus(); + setConnectedChannels(statuses.filter(s => s.connected)); + } catch (err) { + console.debug('[MemoryWorkspace] listStatus failed (non-fatal):', err); + setConnectedChannels([]); + } + setMemoryDocs(docs); setMemoryNamespaces(namespacesPayload); setMemoryFilesList(combinedFiles); @@ -309,6 +336,84 @@ export function MemoryWorkspace({ onToast }: MemoryWorkspaceProps) { } }, [loadSelectedFile, loadWorkspace, memoryNote, onToast, refetchStats]); + // --------------------------------------------------------------------------- + // Sync handlers + // --------------------------------------------------------------------------- + + const handleSyncAll = useCallback(async () => { + setSyncingAll(true); + try { + await memorySyncAll(); + onToast({ + type: 'success', + title: 'Sync Requested', + message: 'Sync requested for all channels.', + }); + } catch (err) { + onToast({ + type: 'error', + title: 'Sync Failed', + message: err instanceof Error ? err.message : 'Sync all failed.', + }); + } finally { + setSyncingAll(false); + } + }, [onToast]); + + const handleSyncChannel = useCallback( + async (channelId: string) => { + setSyncingChannelId(channelId); + try { + await memorySyncChannel(channelId); + onToast({ + type: 'success', + title: 'Sync Requested', + message: `Sync requested for channel ${channelId}.`, + }); + } catch (err) { + onToast({ + type: 'error', + title: 'Sync Failed', + message: err instanceof Error ? err.message : 'Sync failed.', + }); + } finally { + setSyncingChannelId(null); + } + }, + [onToast] + ); + + // --------------------------------------------------------------------------- + // Learn handler + // --------------------------------------------------------------------------- + + const handleLearnAll = useCallback(async () => { + setLearning(true); + setLearnResult(null); + setLearnErrorOpen(false); + try { + const result = await memoryLearnAll(); + setLearnResult(result); + onToast({ + type: 'success', + title: 'Learn Complete', + message: `${result.namespaces_processed} namespace(s) processed.`, + }); + // Refresh workspace data after a successful learn; errors here are + // non-fatal (learn already completed) so we don't surface them as + // "Learn Failed". + void Promise.allSettled([loadWorkspace(), refetchStats()]); + } catch (err) { + onToast({ + type: 'error', + title: 'Learn Failed', + message: err instanceof Error ? err.message : 'Learn failed.', + }); + } finally { + setLearning(false); + } + }, [loadWorkspace, onToast, refetchStats]); + // --------------------------------------------------------------------------- // Derived data // --------------------------------------------------------------------------- @@ -565,6 +670,163 @@ export function MemoryWorkspace({ onToast }: MemoryWorkspaceProps) { )} + {/* Collapsible: Sync */} +
+ + + {syncOpen && ( +
+
+

+ Request ingestion from all connected channels. +

+ +
+ + {connectedChannels.length === 0 ? ( +

No connected channels found.

+ ) : ( +
+ {connectedChannels.map(ch => ( +
+
+
+ {ch.channel_id} +
+
{ch.auth_mode}
+
+ +
+ ))} +
+ )} +
+ )} +
+ + {/* Collapsible: Learn */} +
+ + + {learnOpen && ( +
+

+ Runs the summarizer tree over raw memory namespaces, condensing buffered content into + hour-level summaries and propagating them upward. Requires local AI to be enabled. +

+ + + + {learnResult && ( +
+
+ + {learnResult.namespaces_processed} processed + + · + + {learnResult.results.filter(r => r.status === 'ok').length} ok + + {learnResult.results.some(r => r.status === 'error') && ( + <> + · + + + )} +
+ + {learnErrorOpen && ( +
+ {learnResult.results + .filter(r => r.status === 'error') + .map(r => ( +
+ {r.namespace}:{' '} + {r.error ?? 'unknown error'} +
+ ))} +
+ )} +
+ )} +
+ )} +
+ {/* Warnings */} {(memoryWorkspaceError || (!isTauri() && !memoryWorkspaceLoading)) && (
diff --git a/app/src/components/intelligence/__tests__/MemoryWorkspace.test.tsx b/app/src/components/intelligence/__tests__/MemoryWorkspace.test.tsx index c94915f9f..f4fe539da 100644 --- a/app/src/components/intelligence/__tests__/MemoryWorkspace.test.tsx +++ b/app/src/components/intelligence/__tests__/MemoryWorkspace.test.tsx @@ -1,4 +1,4 @@ -import { screen, waitFor } from '@testing-library/react'; +import { fireEvent, screen, waitFor } from '@testing-library/react'; import { describe, expect, it, vi } from 'vitest'; import { renderWithProviders } from '../../../test/test-utils'; @@ -15,6 +15,21 @@ vi.mock('../../../hooks/useIntelligenceStats', () => ({ }), })); +// Mock channelConnectionsApi so listStatus doesn't hit the network +vi.mock('../../../services/api/channelConnectionsApi', () => ({ + channelConnectionsApi: { + listStatus: vi.fn().mockResolvedValue([ + { + channel_id: 'telegram-main', + auth_mode: 'managed_dm', + connected: true, + has_credentials: true, + }, + { channel_id: 'discord-bot', auth_mode: 'bot_token', connected: true, has_credentials: true }, + ]), + }, +})); + // Override the global tauriCommands mock from setup.ts with memory-specific stubs vi.mock('../../../utils/tauriCommands', () => ({ isTauri: vi.fn(() => true), @@ -31,6 +46,15 @@ vi.mock('../../../utils/tauriCommands', () => ({ memoryDeleteDocument: vi.fn().mockResolvedValue(undefined), memoryQueryNamespace: vi.fn().mockResolvedValue({ text: 'query result', entities: [] }), memoryRecallNamespace: vi.fn().mockResolvedValue({ text: 'recall result', entities: [] }), + memorySyncAll: vi.fn().mockResolvedValue({ requested: true }), + memorySyncChannel: vi.fn().mockResolvedValue({ requested: true, channel_id: 'telegram-main' }), + memoryLearnAll: vi.fn().mockResolvedValue({ + namespaces_processed: 2, + results: [ + { namespace: 'research', status: 'ok' }, + { namespace: 'conversations', status: 'ok' }, + ], + }), memoryGraphQuery: vi.fn().mockResolvedValue([ { namespace: 'research', @@ -146,3 +170,88 @@ describe('MemoryWorkspace – non-Tauri environment', () => { vi.mocked(tauriMod.isTauri).mockReturnValue(true); }); }); + +describe('MemoryWorkspace – Sync section', () => { + const onToast = vi.fn(); + + it('renders the Sync collapsible button', async () => { + renderWithProviders(); + await waitFor(() => { + expect(screen.getByText('Sync')).toBeInTheDocument(); + }); + }); + + it('expands and shows Sync all button when toggled', async () => { + renderWithProviders(); + await waitFor(() => { + expect(screen.getByText('Sync')).toBeInTheDocument(); + }); + fireEvent.click(screen.getByText('Sync').closest('button')!); + await waitFor(() => { + expect(screen.getByText('Sync all')).toBeInTheDocument(); + }); + }); + + it('calls memorySyncAll when Sync all button clicked', async () => { + const tauriMod = await import('../../../utils/tauriCommands'); + renderWithProviders(); + await waitFor(() => screen.getByText('Sync')); + fireEvent.click(screen.getByText('Sync').closest('button')!); + await waitFor(() => screen.getByText('Sync all')); + fireEvent.click(screen.getByText('Sync all')); + await waitFor(() => { + expect(vi.mocked(tauriMod.memorySyncAll)).toHaveBeenCalled(); + }); + }); + + it('calls memorySyncChannel when per-channel Sync button clicked', async () => { + const tauriMod = await import('../../../utils/tauriCommands'); + renderWithProviders(); + await waitFor(() => screen.getByText('Sync')); + fireEvent.click(screen.getByText('Sync').closest('button')!); + await waitFor(() => screen.getAllByText('Sync').length > 1); + // The per-channel sync buttons appear after channels load + await waitFor(() => screen.getByText('telegram-main')); + const syncBtns = screen.getAllByText('Sync'); + // Last Sync buttons are per-channel (first is the header) + fireEvent.click(syncBtns[syncBtns.length - 1]); + await waitFor(() => { + expect(vi.mocked(tauriMod.memorySyncChannel)).toHaveBeenCalledWith('discord-bot'); + }); + }); +}); + +describe('MemoryWorkspace – Learn section', () => { + const onToast = vi.fn(); + + it('renders the Learn collapsible button', async () => { + renderWithProviders(); + await waitFor(() => { + expect(screen.getByText('Learn')).toBeInTheDocument(); + }); + }); + + it('expands and shows Learn all button when toggled', async () => { + renderWithProviders(); + await waitFor(() => screen.getByText('Learn')); + fireEvent.click(screen.getByText('Learn').closest('button')!); + await waitFor(() => { + expect(screen.getByText('Learn all')).toBeInTheDocument(); + }); + }); + + it('calls memoryLearnAll and shows result summary', async () => { + const tauriMod = await import('../../../utils/tauriCommands'); + renderWithProviders(); + await waitFor(() => screen.getByText('Learn')); + fireEvent.click(screen.getByText('Learn').closest('button')!); + await waitFor(() => screen.getByText('Learn all')); + fireEvent.click(screen.getByText('Learn all')); + await waitFor(() => { + expect(vi.mocked(tauriMod.memoryLearnAll)).toHaveBeenCalled(); + }); + await waitFor(() => { + expect(screen.getByText(/2 processed/)).toBeInTheDocument(); + }); + }); +}); diff --git a/app/src/utils/tauriCommands/memory.test.ts b/app/src/utils/tauriCommands/memory.test.ts new file mode 100644 index 000000000..bd54af812 --- /dev/null +++ b/app/src/utils/tauriCommands/memory.test.ts @@ -0,0 +1,111 @@ +/** + * Unit tests for memory RPC wrappers: memorySyncChannel, memorySyncAll, memoryLearnAll. + */ +import { beforeEach, describe, expect, type Mock, test, vi } from 'vitest'; + +import { callCoreRpc } from '../../services/coreRpcClient'; +import { isTauri } from './common'; +import { memoryLearnAll, memorySyncAll, memorySyncChannel } from './memory'; + +vi.mock('../../services/coreRpcClient', () => ({ callCoreRpc: vi.fn() })); +vi.mock('./common', () => ({ isTauri: vi.fn(() => true) })); + +const mockCallCoreRpc = callCoreRpc as Mock; +const mockIsTauri = isTauri as Mock; + +beforeEach(() => { + vi.clearAllMocks(); + mockIsTauri.mockReturnValue(true); +}); + +describe('memorySyncChannel', () => { + test('throws when not in Tauri', async () => { + mockIsTauri.mockReturnValue(false); + await expect(memorySyncChannel('ch-1')).rejects.toThrow('Not running in Tauri'); + expect(mockCallCoreRpc).not.toHaveBeenCalled(); + }); + + test('calls memory_sync_channel with correct channel_id', async () => { + const mockResp = { requested: true, channel_id: 'ch-1' }; + mockCallCoreRpc.mockResolvedValueOnce(mockResp); + + const result = await memorySyncChannel('ch-1'); + + expect(mockCallCoreRpc).toHaveBeenCalledWith({ + method: 'openhuman.memory_sync_channel', + params: { channel_id: 'ch-1' }, + }); + expect(result).toEqual(mockResp); + }); +}); + +describe('memorySyncAll', () => { + test('throws when not in Tauri', async () => { + mockIsTauri.mockReturnValue(false); + await expect(memorySyncAll()).rejects.toThrow('Not running in Tauri'); + }); + + test('calls memory_sync_all and returns result', async () => { + mockCallCoreRpc.mockResolvedValueOnce({ requested: true }); + + const result = await memorySyncAll(); + + expect(mockCallCoreRpc).toHaveBeenCalledWith({ method: 'openhuman.memory_sync_all' }); + expect(result).toEqual({ requested: true }); + }); +}); + +describe('memoryLearnAll', () => { + test('throws when not in Tauri', async () => { + mockIsTauri.mockReturnValue(false); + await expect(memoryLearnAll()).rejects.toThrow('Not running in Tauri'); + }); + + test('calls memory_learn_all without namespaces param when none provided', async () => { + const mockResp = { namespaces_processed: 0, results: [] }; + mockCallCoreRpc.mockResolvedValueOnce(mockResp); + + const result = await memoryLearnAll(); + + expect(mockCallCoreRpc).toHaveBeenCalledWith({ + method: 'openhuman.memory_learn_all', + params: {}, + }); + expect(result.namespaces_processed).toBe(0); + }); + + test('includes namespaces param when provided', async () => { + const mockResp = { + namespaces_processed: 1, + results: [{ namespace: 'research', status: 'ok' }], + }; + mockCallCoreRpc.mockResolvedValueOnce(mockResp); + + const result = await memoryLearnAll(['research']); + + expect(mockCallCoreRpc).toHaveBeenCalledWith({ + method: 'openhuman.memory_learn_all', + params: { namespaces: ['research'] }, + }); + expect(result.namespaces_processed).toBe(1); + expect(result.results[0].status).toBe('ok'); + }); + + test('returns aggregated error results without throwing', async () => { + const mockResp = { + namespaces_processed: 2, + results: [ + { namespace: 'ns-a', status: 'ok' }, + { namespace: 'ns-b', status: 'error', error: 'local AI not enabled' }, + ], + }; + mockCallCoreRpc.mockResolvedValueOnce(mockResp); + + const result = await memoryLearnAll(); + + expect(result.namespaces_processed).toBe(2); + const errEntry = result.results.find(r => r.status === 'error'); + expect(errEntry?.namespace).toBe('ns-b'); + expect(errEntry?.error).toContain('local AI'); + }); +}); diff --git a/app/src/utils/tauriCommands/memory.ts b/app/src/utils/tauriCommands/memory.ts index 40c428b16..02aa348b9 100644 --- a/app/src/utils/tauriCommands/memory.ts +++ b/app/src/utils/tauriCommands/memory.ts @@ -267,3 +267,76 @@ export async function aiWriteMemoryFile(relativePath: string, content: string): params: { relative_path: relativePath, content }, }); } + +export interface MemorySyncChannelResult { + requested: boolean; + channel_id: string; +} + +export interface MemorySyncAllResult { + requested: boolean; +} + +export interface NamespaceLearnResult { + namespace: string; + status: 'ok' | 'skipped' | 'error'; + error?: string; +} + +export interface MemoryLearnAllResult { + namespaces_processed: number; + results: NamespaceLearnResult[]; +} + +/** + * Request a memory sync for a specific channel. + * Publishes MemorySyncRequested on the core event bus and returns confirmation. + * No ingestion runs synchronously — future subscribers will react. + */ +export async function memorySyncChannel(channelId: string): Promise { + console.debug('[memory.sync] memorySyncChannel: entry channel_id=%s', channelId); + if (!isTauri()) { + throw new Error('Not running in Tauri'); + } + const resp = await callCoreRpc({ + method: 'openhuman.memory_sync_channel', + params: { channel_id: channelId }, + }); + console.debug('[memory.sync] memorySyncChannel: exit result=%o', resp); + return resp; +} + +/** + * Request a memory sync for all channels. + * Publishes MemorySyncRequested { channel_id: None } on the core event bus. + */ +export async function memorySyncAll(): Promise { + console.debug('[memory.sync] memorySyncAll: entry'); + if (!isTauri()) { + throw new Error('Not running in Tauri'); + } + const resp = await callCoreRpc({ method: 'openhuman.memory_sync_all' }); + console.debug('[memory.sync] memorySyncAll: exit result=%o', resp); + return resp; +} + +/** + * Run the tree summarizer over all memory namespaces (or a subset). + * Processes sequentially; a failing namespace is recorded, not fatal. + */ +export async function memoryLearnAll(namespaces?: string[]): Promise { + console.debug('[memory.learn] memoryLearnAll: entry namespaces=%o', namespaces); + if (!isTauri()) { + throw new Error('Not running in Tauri'); + } + const params: Record = {}; + if (namespaces && namespaces.length > 0) { + params.namespaces = namespaces; + } + const resp = await callCoreRpc({ + method: 'openhuman.memory_learn_all', + params, + }); + console.debug('[memory.learn] memoryLearnAll: exit processed=%d', resp?.namespaces_processed); + return resp; +} diff --git a/src/core/event_bus/events.rs b/src/core/event_bus/events.rs index 1f6cdf50c..7610eadb7 100644 --- a/src/core/event_bus/events.rs +++ b/src/core/event_bus/events.rs @@ -65,6 +65,13 @@ pub enum DomainEvent { }, /// A memory recall query completed. MemoryRecalled { query: String, hit_count: usize }, + /// A memory sync was requested for a specific channel or all channels. + /// + /// Published by `openhuman.memory_sync_channel` (channel_id = Some(...)) and + /// `openhuman.memory_sync_all` (channel_id = None). No consumers exist yet — + /// this variant is a hook for future ingestion subscribers to react to pull + /// requests. See `src/openhuman/memory/ops.rs` for the RPC handlers. + MemorySyncRequested { channel_id: Option }, // ── Channels ──────────────────────────────────────────────────────── /// An inbound channel message from the transport layer, ready for processing. @@ -357,7 +364,9 @@ impl DomainEvent { | Self::SubagentCompleted { .. } | Self::SubagentFailed { .. } => "agent", - Self::MemoryStored { .. } | Self::MemoryRecalled { .. } => "memory", + Self::MemoryStored { .. } + | Self::MemoryRecalled { .. } + | Self::MemorySyncRequested { .. } => "memory", Self::ChannelInboundMessage { .. } | Self::ChannelMessageReceived { .. } diff --git a/src/openhuman/config/schema/load_tests.rs b/src/openhuman/config/schema/load_tests.rs index 11f6a6faa..677fba46c 100644 --- a/src/openhuman/config/schema/load_tests.rs +++ b/src/openhuman/config/schema/load_tests.rs @@ -832,21 +832,41 @@ async fn resolve_runtime_config_dirs_with_empty_env_falls_back_to_default() { #[test] fn apply_env_overrides_commits_side_effects_to_runtime_proxy() { - use crate::openhuman::config::schema::proxy::runtime_proxy_config; + use crate::openhuman::config::schema::proxy::{runtime_proxy_config, set_runtime_proxy_config}; - // Build a config with a proxy URL via the injectable seam. + // Hold the env lock so no other test races on proxy-related env vars. + let _g = ENV_LOCK.lock().unwrap(); + clear_env(&[ + "OPENHUMAN_PROXY_ENABLED", + "OPENHUMAN_HTTP_PROXY", + "HTTP_PROXY", + "OPENHUMAN_HTTPS_PROXY", + "HTTPS_PROXY", + "OPENHUMAN_ALL_PROXY", + "ALL_PROXY", + ]); + + // Snapshot the global runtime proxy config so we can restore it afterwards + // and avoid leaking state into other tests. + let previous_runtime = runtime_proxy_config(); + + // Build a config with proxy fields set directly on the struct. + // We cannot pre-configure via apply_env_overlay_with + a HashMapEnv and + // then call apply_env_overrides(), because apply_env_overrides() internally + // re-runs apply_env_overlay_with(&ProcessEnv) which reads the real process + // environment — overwriting anything set via a HashMapEnv beforehand. + // Setting fields directly ensures they survive the ProcessEnv overlay + // (which only writes fields when the corresponding env var is present). let mut cfg = Config::default(); - cfg.apply_env_overlay_with( - &HashMapEnv::new() - .with("OPENHUMAN_HTTP_PROXY", "http://proxy.test:8080") - .with("OPENHUMAN_PROXY_ENABLED", "true"), - ); + cfg.proxy.http_proxy = Some("http://proxy.test:8080".to_string()); + cfg.proxy.enabled = true; - // Now call the public wrapper which commits side effects. + // apply_env_overrides commits side effects: it calls set_runtime_proxy_config + // with the current proxy config after the ProcessEnv overlay. cfg.apply_env_overrides(); // `set_runtime_proxy_config` must have been called: the global should - // reflect the proxy URL we injected. + // reflect the proxy URL we set on cfg.proxy. let runtime = runtime_proxy_config(); assert!( runtime.enabled, @@ -855,6 +875,10 @@ fn apply_env_overrides_commits_side_effects_to_runtime_proxy() { assert_eq!( runtime.http_proxy.as_deref(), Some("http://proxy.test:8080"), - "runtime proxy URL must match the env-injected value" + "runtime proxy URL must match the value set on cfg.proxy" ); + + // Restore the global runtime proxy state so this test doesn't bleed into + // other tests that inspect runtime_proxy_config(). + set_runtime_proxy_config(previous_runtime); } diff --git a/src/openhuman/memory/ops.rs b/src/openhuman/memory/ops.rs index f80e23281..1d9770b92 100644 --- a/src/openhuman/memory/ops.rs +++ b/src/openhuman/memory/ops.rs @@ -883,6 +883,216 @@ pub async fn graph_query( Ok(RpcOutcome::single_log(rows, "memory graph queried")) } +/// Parameters for `memory_sync_channel`. +#[derive(Debug, serde::Deserialize)] +pub struct SyncChannelParams { + pub channel_id: String, +} + +/// Result returned by `memory_sync_channel`. +#[derive(Debug, serde::Serialize)] +pub struct SyncChannelResult { + pub requested: bool, + pub channel_id: String, +} + +/// Result returned by `memory_sync_all`. +#[derive(Debug, serde::Serialize)] +pub struct SyncAllResult { + pub requested: bool, +} + +/// Per-namespace outcome for `memory_learn_all`. +#[derive(Debug, serde::Serialize)] +pub struct NamespaceLearnResult { + pub namespace: String, + pub status: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub error: Option, +} + +/// Result returned by `memory_learn_all`. +#[derive(Debug, serde::Serialize)] +pub struct LearnAllResult { + pub namespaces_processed: usize, + pub results: Vec, +} + +/// Parameters for `memory_learn_all`. +#[derive(Debug, serde::Deserialize)] +pub struct LearnAllParams { + /// Optional list of namespaces to constrain. Defaults to all namespaces. + #[serde(default)] + pub namespaces: Option>, +} + +/// Request a memory sync for a specific channel. +/// +/// Ingestion in OpenHuman is listener/webhook-driven — there is no per-provider +/// pull mechanism yet. This RPC publishes `DomainEvent::MemorySyncRequested` so +/// that future ingestion subscribers can react to an explicit pull request. +/// The event is fire-and-forget; the caller receives confirmation that the +/// request was published, not that ingestion ran. +pub async fn memory_sync_channel( + params: SyncChannelParams, +) -> Result, String> { + tracing::info!( + "[memory.sync] memory_sync_channel: entry channel_id={}", + params.channel_id + ); + crate::core::event_bus::publish_global( + crate::core::event_bus::DomainEvent::MemorySyncRequested { + channel_id: Some(params.channel_id.clone()), + }, + ); + tracing::debug!( + "[memory.sync] memory_sync_channel: MemorySyncRequested published channel_id={}", + params.channel_id + ); + Ok(RpcOutcome::new( + SyncChannelResult { + requested: true, + channel_id: params.channel_id, + }, + vec![], + )) +} + +/// Request a memory sync for all channels. +/// +/// Publishes `DomainEvent::MemorySyncRequested { channel_id: None }` on the +/// global event bus. No consumers exist yet — this is a hook for future +/// ingestion subscribers. +pub async fn memory_sync_all() -> Result, String> { + tracing::info!("[memory.sync] memory_sync_all: entry"); + crate::core::event_bus::publish_global( + crate::core::event_bus::DomainEvent::MemorySyncRequested { channel_id: None }, + ); + tracing::debug!("[memory.sync] memory_sync_all: MemorySyncRequested(all) published"); + Ok(RpcOutcome::new(SyncAllResult { requested: true }, vec![])) +} + +/// Run the tree summarizer over all (or a constrained set of) namespaces. +/// +/// Enumerates namespaces via `namespace_list`, then for each runs +/// `tree_summarizer_run`. Results are collected per-namespace; a failing +/// namespace does not abort the rest. Runs sequentially to avoid saturating +/// the local AI provider. +pub async fn memory_learn_all( + params: LearnAllParams, +) -> Result, String> { + tracing::info!( + "[memory.learn] memory_learn_all: entry namespaces={:?}", + params.namespaces + ); + + // Resolve the target namespace list. + let client = active_memory_client().await?; + let all_ns = client.list_namespaces().await?; + tracing::debug!("[memory.learn] available namespaces: {:?}", all_ns); + + let target_ns: Vec = match ¶ms.namespaces { + Some(requested) if !requested.is_empty() => { + let mut seen = BTreeSet::new(); + let filtered: Vec<_> = requested + .iter() + .filter(|ns| all_ns.contains(ns)) + .filter(|ns| seen.insert((*ns).clone())) + .cloned() + .collect(); + tracing::debug!("[memory.learn] constrained to namespaces: {:?}", filtered); + filtered + } + Some(requested) => { + // Explicit empty list → no-op (don't fall back to all namespaces). + let mut seen = BTreeSet::new(); + let filtered: Vec<_> = requested + .iter() + .filter(|ns| all_ns.contains(ns)) + .filter(|ns| seen.insert((*ns).clone())) + .cloned() + .collect(); + tracing::debug!( + "[memory.learn] Some([]) empty request → no-op or filtered to {:?}", + filtered + ); + filtered + } + None => { + tracing::debug!("[memory.learn] using all {} namespaces", all_ns.len()); + all_ns + } + }; + + // Short-circuit when there are no namespaces to process — avoids loading + // config (and the local_ai.enabled guard) for an empty batch. + if target_ns.is_empty() { + tracing::info!( + "[memory.learn] memory_learn_all: no namespaces to process, returning early" + ); + return Ok(RpcOutcome::new( + LearnAllResult { + namespaces_processed: 0, + results: vec![], + }, + vec![], + )); + } + + let config = crate::openhuman::config::Config::load_or_init() + .await + .map_err(|e| format!("load config: {e}"))?; + + if !config.local_ai.enabled { + return Err("memory_learn_all requires local_ai.enabled=true".to_string()); + } + + let mut results = Vec::with_capacity(target_ns.len()); + for namespace in &target_ns { + tracing::info!( + "[memory.learn] running summarization for namespace='{}'", + namespace + ); + let outcome = + crate::openhuman::tree_summarizer::ops::tree_summarizer_run(&config, namespace).await; + match outcome { + Ok(_) => { + tracing::info!("[memory.learn] namespace='{}' ok", namespace); + results.push(NamespaceLearnResult { + namespace: namespace.clone(), + status: "ok".to_string(), + error: None, + }); + } + Err(e) => { + tracing::warn!("[memory.learn] namespace='{}' error: {}", namespace, e); + results.push(NamespaceLearnResult { + namespace: namespace.clone(), + status: "error".to_string(), + error: Some(e), + }); + } + } + } + + let namespaces_processed = results.len(); + tracing::info!( + "[memory.learn] memory_learn_all: done processed={} results={:?}", + namespaces_processed, + results + .iter() + .map(|r| (&r.namespace, &r.status)) + .collect::>() + ); + Ok(RpcOutcome::new( + LearnAllResult { + namespaces_processed, + results, + }, + vec![], + )) +} + /// Initialise the local-only (SQLite) memory subsystem for the current workspace. /// /// `request.jwt_token` is accepted for backward compatibility but ignored — all diff --git a/src/openhuman/memory/schemas.rs b/src/openhuman/memory/schemas.rs index 45b579f6a..8e8ac2b6d 100644 --- a/src/openhuman/memory/schemas.rs +++ b/src/openhuman/memory/schemas.rs @@ -11,8 +11,8 @@ use crate::core::all::{ControllerFuture, RegisteredController}; use crate::core::{ControllerSchema, FieldSchema, TypeSchema}; use crate::openhuman::memory::rpc::{ self, ClearNamespaceParams, DeleteDocParams, GraphQueryParams, GraphUpsertParams, - IngestDocParams, KvGetDeleteParams, KvSetParams, NamespaceOnlyParams, PutDocParams, - QueryNamespaceParams, RecallNamespaceParams, + IngestDocParams, KvGetDeleteParams, KvSetParams, LearnAllParams, NamespaceOnlyParams, + PutDocParams, QueryNamespaceParams, RecallNamespaceParams, SyncChannelParams, }; use crate::openhuman::memory::{ DeleteDocumentRequest, EmptyRequest, ListDocumentsRequest, ListMemoryFilesRequest, @@ -52,6 +52,9 @@ pub fn all_controller_schemas() -> Vec { schemas("graph_upsert"), schemas("graph_query"), schemas("clear_namespace"), + schemas("sync_channel"), + schemas("sync_all"), + schemas("learn_all"), ] } @@ -154,6 +157,18 @@ pub fn all_registered_controllers() -> Vec { schema: schemas("clear_namespace"), handler: handle_clear_namespace, }, + RegisteredController { + schema: schemas("sync_channel"), + handler: handle_sync_channel, + }, + RegisteredController { + schema: schemas("sync_all"), + handler: handle_sync_all, + }, + RegisteredController { + schema: schemas("learn_all"), + handler: handle_learn_all, + }, ] } @@ -914,6 +929,70 @@ pub fn schemas(function: &str) -> ControllerSchema { ], }, + // ----- sync / learn ----- + "sync_channel" => ControllerSchema { + namespace: "memory", + function: "sync_channel", + description: "Request a memory sync for a specific channel. Publishes MemorySyncRequested on the event bus. No ingestion consumers exist yet; this is a hook for future pull-based subscribers.", + inputs: vec![FieldSchema { + name: "channel_id", + ty: TypeSchema::String, + comment: "ID of the channel to sync.", + required: true, + }], + outputs: vec![ + FieldSchema { + name: "requested", + ty: TypeSchema::Bool, + comment: "Always true when the event was published.", + required: true, + }, + FieldSchema { + name: "channel_id", + ty: TypeSchema::String, + comment: "Echo of the channel_id that was requested.", + required: true, + }, + ], + }, + "sync_all" => ControllerSchema { + namespace: "memory", + function: "sync_all", + description: "Request a memory sync for all channels. Publishes MemorySyncRequested { channel_id: None } on the event bus.", + inputs: vec![], + outputs: vec![FieldSchema { + name: "requested", + ty: TypeSchema::Bool, + comment: "Always true when the event was published.", + required: true, + }], + }, + "learn_all" => ControllerSchema { + namespace: "memory", + function: "learn_all", + description: "Run the tree summarizer over all memory namespaces (or a constrained subset). Processes namespaces sequentially; a failing namespace is recorded but does not abort the rest.", + inputs: vec![FieldSchema { + name: "namespaces", + ty: TypeSchema::Option(Box::new(TypeSchema::Array(Box::new(TypeSchema::String)))), + comment: "Optional list of namespaces to constrain. Defaults to all namespaces.", + required: false, + }], + outputs: vec![ + FieldSchema { + name: "namespaces_processed", + ty: TypeSchema::U64, + comment: "Total number of namespaces processed.", + required: true, + }, + FieldSchema { + name: "results", + ty: TypeSchema::Json, + comment: "Per-namespace outcomes: [{ namespace, status: 'ok'|'skipped'|'error', error? }].", + required: true, + }, + ], + }, + // ----- fallback ----- _other => ControllerSchema { namespace: "memory", @@ -1112,6 +1191,24 @@ fn handle_clear_namespace(params: Map) -> ControllerFuture { }) } +fn handle_sync_channel(params: Map) -> ControllerFuture { + Box::pin(async move { + let payload = parse_params::(params)?; + to_json(rpc::memory_sync_channel(payload).await?) + }) +} + +fn handle_sync_all(_params: Map) -> ControllerFuture { + Box::pin(async move { to_json(rpc::memory_sync_all().await?) }) +} + +fn handle_learn_all(params: Map) -> ControllerFuture { + Box::pin(async move { + let payload = parse_params::(params)?; + to_json(rpc::memory_learn_all(payload).await?) + }) +} + // --------------------------------------------------------------------------- // Helpers // --------------------------------------------------------------------------- diff --git a/src/openhuman/memory/schemas_tests.rs b/src/openhuman/memory/schemas_tests.rs index f2d5c0edb..605fe895a 100644 --- a/src/openhuman/memory/schemas_tests.rs +++ b/src/openhuman/memory/schemas_tests.rs @@ -26,6 +26,9 @@ const ALL_FUNCTIONS: &[&str] = &[ "graph_upsert", "graph_query", "clear_namespace", + "sync_channel", + "sync_all", + "learn_all", ]; #[test] @@ -108,3 +111,39 @@ fn parse_params_surfaces_deserialization_errors_with_context() { let err = parse_params::(m).unwrap_err(); assert!(err.contains("invalid params")); } + +// ── sync / learn schema shape tests ───────────────────────────────────── + +#[test] +fn sync_channel_schema_requires_channel_id() { + let s = schemas("sync_channel"); + assert_eq!(s.namespace, "memory"); + assert_eq!(s.function, "sync_channel"); + let required: Vec<_> = s + .inputs + .iter() + .filter(|f| f.required) + .map(|f| f.name) + .collect(); + assert!( + required.contains(&"channel_id"), + "channel_id must be required" + ); +} + +#[test] +fn sync_all_schema_has_no_inputs() { + let s = schemas("sync_all"); + assert_eq!(s.function, "sync_all"); + assert!(s.inputs.is_empty(), "sync_all takes no inputs"); +} + +#[test] +fn learn_all_schema_namespaces_is_optional() { + let s = schemas("learn_all"); + assert_eq!(s.function, "learn_all"); + assert_eq!(s.inputs.len(), 1); + let ns_field = &s.inputs[0]; + assert_eq!(ns_field.name, "namespaces"); + assert!(!ns_field.required, "namespaces must be optional"); +} diff --git a/tests/json_rpc_e2e.rs b/tests/json_rpc_e2e.rs index 09689ac15..93f3a7fd4 100644 --- a/tests/json_rpc_e2e.rs +++ b/tests/json_rpc_e2e.rs @@ -901,6 +901,105 @@ async fn json_rpc_thread_labels_create_and_update() { rpc_join.abort(); } +#[tokio::test] +async fn json_rpc_memory_sync_and_learn() { + let _env_lock = json_rpc_e2e_env_lock(); + let tmp = tempdir().expect("tempdir"); + let home = tmp.path(); + let openhuman_home = home.join(".openhuman"); + + let _home_guard = EnvVarGuard::set_to_path("HOME", home); + let _workspace_guard = EnvVarGuard::unset("OPENHUMAN_WORKSPACE"); + let _backend_url_guard = EnvVarGuard::unset("BACKEND_URL"); + let _vite_backend_guard = EnvVarGuard::unset("VITE_BACKEND_URL"); + let _embed_strict_guard = EnvVarGuard::set("OPENHUMAN_MEMORY_EMBED_STRICT", "false"); + let _embed_endpoint_guard = EnvVarGuard::set("OPENHUMAN_MEMORY_EMBED_ENDPOINT", ""); + let _embed_model_guard = EnvVarGuard::set("OPENHUMAN_MEMORY_EMBED_MODEL", ""); + + let (mock_addr, mock_join) = serve_on_ephemeral(mock_upstream_router()).await; + let mock_origin = format!("http://{}", mock_addr); + write_min_config(&openhuman_home, &mock_origin); + + let (rpc_addr, rpc_join) = serve_on_ephemeral(build_core_http_router(false)).await; + let rpc_base = format!("http://{rpc_addr}"); + + tokio::time::sleep(Duration::from_millis(100)).await; + + // ── memory_sync_all: returns requested:true ────────────────────────────── + let sync_all = post_json_rpc(&rpc_base, 7001, "openhuman.memory_sync_all", json!({})).await; + let sync_all_result = assert_no_jsonrpc_error(&sync_all, "memory_sync_all"); + assert_eq!( + sync_all_result.get("requested"), + Some(&json!(true)), + "memory_sync_all must return requested:true" + ); + + // ── memory_sync_channel: echoes channel_id and returns requested:true ───── + let sync_ch = post_json_rpc( + &rpc_base, + 7002, + "openhuman.memory_sync_channel", + json!({ "channel_id": "test-channel-abc" }), + ) + .await; + let sync_ch_result = assert_no_jsonrpc_error(&sync_ch, "memory_sync_channel"); + assert_eq!( + sync_ch_result.get("requested"), + Some(&json!(true)), + "memory_sync_channel must return requested:true" + ); + assert_eq!( + sync_ch_result.get("channel_id").and_then(Value::as_str), + Some("test-channel-abc"), + "memory_sync_channel must echo channel_id" + ); + + // ── memory_sync_channel: missing channel_id returns a JSON-RPC error ──── + let sync_bad = post_json_rpc(&rpc_base, 7003, "openhuman.memory_sync_channel", json!({})).await; + assert!( + sync_bad.get("error").is_some(), + "missing channel_id must return an error, got: {sync_bad}" + ); + + // ── memory_learn_all: no namespaces → zero processed (empty store) ────── + let learn_all = post_json_rpc(&rpc_base, 7004, "openhuman.memory_learn_all", json!({})).await; + let learn_result = assert_no_jsonrpc_error(&learn_all, "memory_learn_all"); + let processed = learn_result + .get("namespaces_processed") + .and_then(Value::as_u64) + .expect("namespaces_processed must be present"); + assert_eq!(processed, 0, "no namespaces in a fresh store"); + let results_arr = learn_result + .get("results") + .and_then(Value::as_array) + .expect("results array must be present"); + assert!( + results_arr.is_empty(), + "results must be empty when no namespaces" + ); + + // ── memory_learn_all: constrained to non-existent namespace → also zero ── + let learn_constrained = post_json_rpc( + &rpc_base, + 7005, + "openhuman.memory_learn_all", + json!({ "namespaces": ["does-not-exist"] }), + ) + .await; + let learn_c_result = + assert_no_jsonrpc_error(&learn_constrained, "memory_learn_all constrained"); + assert_eq!( + learn_c_result + .get("namespaces_processed") + .and_then(Value::as_u64), + Some(0), + "non-existent namespace must be filtered out" + ); + + mock_join.abort(); + rpc_join.abort(); +} + #[tokio::test] async fn json_rpc_memory_tree_end_to_end() { let _env_lock = json_rpc_e2e_env_lock();