diff --git a/app/src/components/intelligence/MemorySyncConnections.test.tsx b/app/src/components/intelligence/MemorySyncConnections.test.tsx new file mode 100644 index 000000000..c332fa0a5 --- /dev/null +++ b/app/src/components/intelligence/MemorySyncConnections.test.tsx @@ -0,0 +1,75 @@ +import { render, screen, waitFor } from '@testing-library/react'; +import { beforeEach, describe, expect, it, vi } from 'vitest'; + +import type { MemorySyncStatus } from '../../services/memorySyncService'; +import { MemorySyncConnections } from './MemorySyncConnections'; + +const mockStatusList = vi.fn(); + +vi.mock('../../services/memorySyncService', async () => { + const actual = await vi.importActual( + '../../services/memorySyncService' + ); + return { ...actual, memorySyncStatusList: (...args: unknown[]) => mockStatusList(...args) }; +}); + +function makeStatus(overrides: Partial = {}): MemorySyncStatus { + return { + provider: 'gmail', + chunks_synced: 0, + chunks_pending: 0, + batch_total: 0, + batch_processed: 0, + last_chunk_at_ms: null, + freshness: 'idle', + ...overrides, + }; +} + +describe('', () => { + beforeEach(() => { + mockStatusList.mockReset(); + }); + + it('renders a card per provider from the RPC', async () => { + mockStatusList.mockResolvedValueOnce([ + makeStatus({ provider: 'gmail', chunks_synced: 42, freshness: 'active' }), + makeStatus({ provider: 'discord', chunks_synced: 7, freshness: 'recent' }), + ]); + render(); + await waitFor(() => { + expect(screen.getByTestId('memory-sync-card-gmail')).toBeTruthy(); + expect(screen.getByTestId('memory-sync-card-discord')).toBeTruthy(); + }); + expect(screen.getByText('Gmail')).toBeTruthy(); + expect(screen.getByText('Discord')).toBeTruthy(); + }); + + it('shows chunk count + freshness label', async () => { + mockStatusList.mockResolvedValueOnce([ + makeStatus({ provider: 'notion', chunks_synced: 1234, freshness: 'recent' }), + ]); + render(); + await waitFor(() => { + expect(screen.getByTestId('memory-sync-chunks-notion').textContent).toContain('1,234'); + expect(screen.getByTestId('memory-sync-freshness-notion').textContent).toBe('Recent'); + }); + }); + + it('renders the empty state when the RPC returns []', async () => { + mockStatusList.mockResolvedValueOnce([]); + render(); + await waitFor(() => { + expect(screen.getByText(/No content has been synced/)).toBeTruthy(); + }); + }); + + it('renders the failure state when the RPC throws', async () => { + mockStatusList.mockRejectedValueOnce(new Error('rpc unavailable')); + render(); + await waitFor(() => { + expect(screen.getByText(/Failed to load sync status/)).toBeTruthy(); + expect(screen.getByText(/rpc unavailable/)).toBeTruthy(); + }); + }); +}); diff --git a/app/src/components/intelligence/MemorySyncConnections.tsx b/app/src/components/intelligence/MemorySyncConnections.tsx new file mode 100644 index 000000000..1ce79c634 --- /dev/null +++ b/app/src/components/intelligence/MemorySyncConnections.tsx @@ -0,0 +1,218 @@ +/** + * Memory sync card list (#1136 — simplified rewrite). + * + * Renders one card per `source_kind` (data-source type) that has chunks + * in the memory tree. Counts come straight from a SQL aggregate over + * `mem_tree_chunks` so the snapshot is always exact at the moment of + * the poll. No phases, no settings, no per-connection state — chunks + * exist or they don't. + */ +import { useCallback, useEffect, useState } from 'react'; + +import { + type FreshnessLabel, + type MemorySyncStatus, + memorySyncStatusList, +} from '../../services/memorySyncService'; + +interface MemorySyncConnectionsProps { + /** Optional pollIntervalMs — when set, the list refetches periodically. */ + pollIntervalMs?: number; +} + +const FRESHNESS_LABEL: Record = { + active: 'Active', + recent: 'Recent', + idle: 'Idle', +}; + +const PROVIDER_LABEL: Record = { + slack: 'Slack', + discord: 'Discord', + telegram: 'Telegram', + whatsapp: 'WhatsApp', + gmail: 'Gmail', + other_email: 'Email', + notion: 'Notion', + meeting_notes: 'Meeting notes', + drive_docs: 'Drive docs', + // category fallbacks (for chunks without a `:` prefix in source_id) + chat: 'Chat', + email: 'Email', + document: 'Document', +}; + +function freshnessBadgeClass(label: FreshnessLabel): string { + switch (label) { + case 'active': + return 'bg-ocean-100 text-ocean-700'; + case 'recent': + return 'bg-sage-100 text-sage-700'; + case 'idle': + return 'bg-stone-100 text-stone-700'; + } +} + +function relativeTimestamp(epochMs: number | null): string | null { + if (epochMs === null) return null; + const delta = Date.now() - epochMs; + if (delta < 1000) return 'just now'; + const seconds = Math.floor(delta / 1000); + if (seconds < 60) return `${seconds}s ago`; + const minutes = Math.floor(seconds / 60); + if (minutes < 60) return `${minutes}m ago`; + const hours = Math.floor(minutes / 60); + if (hours < 24) return `${hours}h ago`; + const days = Math.floor(hours / 24); + return `${days}d ago`; +} + +interface SourceCardProps { + status: MemorySyncStatus; +} + +function SourceCard({ status }: SourceCardProps) { + const label = PROVIDER_LABEL[status.provider] ?? status.provider; + const lastSync = relativeTimestamp(status.last_chunk_at_ms); + const lifetime = status.chunks_synced; + const pending = status.chunks_pending; + // Progress reflects the *active sync wave* (chunks within the most + // recent ingest cluster), not lifetime, so the bar tracks "how much + // of this sync's ingest has been processed". Hidden once the wave + // is fully drained. + const batchTotal = status.batch_total; + const batchProcessed = status.batch_processed; + const batchPending = batchTotal - batchProcessed; + const pct = batchTotal > 0 ? Math.round((batchProcessed / batchTotal) * 100) : 0; + const showProgress = batchTotal > 0 && batchPending > 0; + + return ( +
+
+
+
+ {label} + + {FRESHNESS_LABEL[status.freshness]} + +
+
+ + {lifetime.toLocaleString()} chunks + + {lastSync && Last chunk {lastSync}} +
+
+
+ + {showProgress && ( +
+
+
+
+
+ + {batchProcessed.toLocaleString()} of {batchTotal.toLocaleString()} processed + + {pending > 0 && ( + · {pending.toLocaleString()} pending + )} +
+
+ )} +
+ ); +} + +export function MemorySyncConnections({ pollIntervalMs }: MemorySyncConnectionsProps) { + const [statuses, setStatuses] = useState([]); + const [loadError, setLoadError] = useState(null); + const [loading, setLoading] = useState(true); + + const loadStatuses = useCallback(async () => { + try { + console.debug('[ui-flow][memory-sync] fetching status list'); + const list = await memorySyncStatusList(); + setStatuses(list); + setLoadError(null); + } catch (err) { + const message = err instanceof Error ? err.message : String(err); + console.error('[ui-flow][memory-sync] status list failed', message); + setLoadError(message); + } finally { + setLoading(false); + } + }, []); + + useEffect(() => { + let cancelled = false; + void (async () => { + await loadStatuses(); + if (cancelled) return; + })(); + return () => { + cancelled = true; + }; + }, [loadStatuses]); + + useEffect(() => { + if (!pollIntervalMs) return undefined; + const id = setInterval(() => { + void loadStatuses(); + }, pollIntervalMs); + return () => clearInterval(id); + }, [pollIntervalMs, loadStatuses]); + + if (loading) { + return ( +
+

Memory sources

+

Loading…

+
+ ); + } + + if (loadError) { + return ( +
+

Memory sources

+

+ Failed to load sync status: {loadError} +

+
+ ); + } + + if (statuses.length === 0) { + return ( +
+

Memory sources

+

+ No content has been synced into memory yet. Connect an integration to start. +

+
+ ); + } + + return ( +
+

Memory sources

+
+ {statuses.map(s => ( + + ))} +
+
+ ); +} diff --git a/app/src/components/intelligence/MemoryWorkspace.tsx b/app/src/components/intelligence/MemoryWorkspace.tsx index 3fe0f585a..daf436532 100644 --- a/app/src/components/intelligence/MemoryWorkspace.tsx +++ b/app/src/components/intelligence/MemoryWorkspace.tsx @@ -38,6 +38,7 @@ import { MemoryChunkDetail } from './MemoryChunkDetail'; import { MemoryEmptyPlaceholder } from './MemoryEmptyPlaceholder'; import { MemoryNavigator, type NavigatorSelection } from './MemoryNavigator'; import { MemoryResultList } from './MemoryResultList'; +import { MemorySyncConnections } from './MemorySyncConnections'; interface MemoryWorkspaceProps { onToast?: (toast: Omit) => void; @@ -214,89 +215,95 @@ export function MemoryWorkspace({ onToast: _onToast }: MemoryWorkspaceProps) { if (isEmpty) { return ( -
- +
+ +
+ +
); } return ( -
- {/* 2-pane base */} - +
+ +
+ {/* 2-pane base */} + -
- -
+
+ +
- {/* Detail overlay — fills the entire workspace card */} - {selectedChunk && ( -
-
+
-
-

- {selectedChunk.source_kind} - {' · '} - {selectedChunk.source_id} -

-
-
+ - + + + -
-
- +
+
+ +
-
- )} -
+ )} +
+
); } diff --git a/app/src/services/memorySyncService.test.ts b/app/src/services/memorySyncService.test.ts new file mode 100644 index 000000000..073d655fe --- /dev/null +++ b/app/src/services/memorySyncService.test.ts @@ -0,0 +1,56 @@ +import { beforeEach, describe, expect, it, vi } from 'vitest'; + +import { type MemorySyncStatus, memorySyncStatusList } from './memorySyncService'; + +const mockCallCoreRpc = vi.fn(); + +vi.mock('./coreRpcClient', () => ({ + callCoreRpc: (...args: unknown[]) => mockCallCoreRpc(...args), +})); + +function makeStatus(overrides: Partial = {}): MemorySyncStatus { + return { + provider: 'gmail', + chunks_synced: 0, + chunks_pending: 0, + batch_total: 0, + batch_processed: 0, + last_chunk_at_ms: null, + freshness: 'idle', + ...overrides, + }; +} + +describe('memorySyncService.memorySyncStatusList', () => { + beforeEach(() => { + mockCallCoreRpc.mockReset(); + }); + + it('calls the correct RPC method without params', async () => { + mockCallCoreRpc.mockResolvedValueOnce({ statuses: [] }); + await memorySyncStatusList(); + expect(mockCallCoreRpc).toHaveBeenCalledWith({ method: 'openhuman.memory_sync_status_list' }); + }); + + it('returns the statuses array from the envelope', async () => { + const status = makeStatus({ provider: 'slack', chunks_synced: 42, freshness: 'active' }); + mockCallCoreRpc.mockResolvedValueOnce({ statuses: [status] }); + const out = await memorySyncStatusList(); + expect(out).toEqual([status]); + }); + + it('propagates RPC errors as thrown errors', async () => { + mockCallCoreRpc.mockRejectedValueOnce(new Error('rpc boom')); + await expect(memorySyncStatusList()).rejects.toThrow('rpc boom'); + }); + + it('throws on malformed response (missing statuses[])', async () => { + mockCallCoreRpc.mockResolvedValueOnce({ wrong: 'shape' }); + await expect(memorySyncStatusList()).rejects.toThrow(/missing statuses/); + }); + + it('throws on null response', async () => { + mockCallCoreRpc.mockResolvedValueOnce(null); + await expect(memorySyncStatusList()).rejects.toThrow(/missing statuses/); + }); +}); diff --git a/app/src/services/memorySyncService.ts b/app/src/services/memorySyncService.ts new file mode 100644 index 000000000..a8550c43a --- /dev/null +++ b/app/src/services/memorySyncService.ts @@ -0,0 +1,65 @@ +/** + * Memory-sync RPC client (#1136 — simplified rewrite). + * + * Wraps `openhuman.memory_sync_status_list` so screens don't have to know + * the wire shape. The Rust handler counts chunks in `mem_tree_chunks` + * GROUPED BY `source_kind` on every call and derives a freshness label + * from the most recent chunk's timestamp — no settings, no phases, no + * persisted KV store. The chunks table is the source of truth. + */ +import debug from 'debug'; + +import { callCoreRpc } from './coreRpcClient'; + +const log = debug('memory-sync'); +const errLog = debug('memory-sync:error'); + +/** Activity freshness derived at the server from the most-recent chunk. */ +export type FreshnessLabel = 'active' | 'recent' | 'idle'; + +/** One row per provider that has chunks in the memory tree. */ +export interface MemorySyncStatus { + /** Specific provider — "slack", "gmail", "discord", "telegram", + * "whatsapp", "notion", "meeting_notes", "drive_docs". Derived + * server-side from each chunk's `source_id` prefix. */ + provider: string; + /** Total chunks ingested for this source_kind. */ + chunks_synced: number; + /** Chunks not yet processed (lifetime). Counts every chunk with + * `embedding IS NULL`, regardless of when it was ingested. */ + chunks_pending: number; + /** Total chunks in the current sync wave (chunks created at-or-after + * the oldest currently-pending chunk). Zero when nothing is in + * flight. */ + batch_total: number; + /** Of `batch_total`, how many have been processed since the wave + * started. Progress fill = `batch_processed / batch_total`. */ + batch_processed: number; + /** Most recent chunk's `timestamp_ms` for this source_kind, or `null`. */ + last_chunk_at_ms: number | null; + /** Server-derived freshness label. */ + freshness: FreshnessLabel; +} + +// `callCoreRpc` returns `json.result` from the JSON-RPC envelope. +interface StatusListResponse { + statuses: MemorySyncStatus[]; +} + +/** List one row per source_kind that has chunks. Ordered server-side by recency. */ +export async function memorySyncStatusList(): Promise { + log('memory_sync_status_list: calling core RPC'); + let resp: StatusListResponse; + try { + resp = await callCoreRpc({ method: 'openhuman.memory_sync_status_list' }); + } catch (err) { + errLog('memory_sync_status_list: RPC failed: %O', err); + throw err; + } + if (!resp || !Array.isArray(resp.statuses)) { + errLog('memory_sync_status_list: malformed response (missing statuses[]): %O', resp); + throw new Error('Invalid response from openhuman.memory_sync_status_list: missing statuses[]'); + } + log('memory_sync_status_list: received %d row(s)', resp.statuses.length); + return resp.statuses; +} diff --git a/src/core/all.rs b/src/core/all.rs index 583edb710..7a5b7d1e4 100644 --- a/src/core/all.rs +++ b/src/core/all.rs @@ -152,6 +152,8 @@ fn build_registered_controllers() -> Vec { controllers.extend(crate::openhuman::memory::all_retrieval_registered_controllers()); // Slack → memory-tree ingestion engine (backfill + poll + 6hr bucket flush) controllers.extend(crate::openhuman::memory::all_slack_ingestion_registered_controllers()); + // Per-connection memory sync status, controls, and progress (#1136) + controllers.extend(crate::openhuman::memory::all_memory_sync_status_registered_controllers()); // Link shortener for long tracking URLs — saves LLM tokens controllers .extend(crate::openhuman::redirect_links::all_redirect_links_registered_controllers()); @@ -229,6 +231,7 @@ fn build_declared_controller_schemas() -> Vec { schemas.extend(crate::openhuman::memory::all_memory_tree_controller_schemas()); schemas.extend(crate::openhuman::memory::all_retrieval_controller_schemas()); schemas.extend(crate::openhuman::memory::all_slack_ingestion_controller_schemas()); + schemas.extend(crate::openhuman::memory::all_memory_sync_status_controller_schemas()); schemas.extend(crate::openhuman::redirect_links::all_redirect_links_controller_schemas()); schemas.extend(crate::openhuman::referral::all_referral_controller_schemas()); schemas.extend(crate::openhuman::billing::all_billing_controller_schemas()); @@ -297,6 +300,9 @@ pub fn namespace_description(namespace: &str) -> Option<&'static str> { "memory_tree" => Some( "Canonical chunk ingestion, provenance capture, and chunk retrieval for source-grounded memory.", ), + "memory_sync" => Some( + "Per-connection memory sync status, user enable toggle, and live progress for the desktop UI.", + ), "redirect_links" => Some( "Shorten long tracking URLs to `openhuman://link/` placeholders (SQLite-backed) to save tokens in prompts, with round-trip rewrite helpers.", ), diff --git a/src/openhuman/memory/mod.rs b/src/openhuman/memory/mod.rs index efaabaa3d..0f66377c0 100644 --- a/src/openhuman/memory/mod.rs +++ b/src/openhuman/memory/mod.rs @@ -15,6 +15,7 @@ pub mod safety; pub mod schemas; pub mod slack_ingestion; pub mod store; +pub mod sync_status; pub mod traits; pub mod tree; @@ -39,6 +40,10 @@ pub use store::{ MemoryClientRef, MemoryItemKind, MemoryState, NamespaceDocumentInput, NamespaceMemoryHit, NamespaceQueryResult, NamespaceRetrievalContext, RetrievalScoreBreakdown, UnifiedMemory, }; +pub use sync_status::{ + all_memory_sync_status_controller_schemas, all_memory_sync_status_registered_controllers, + FreshnessLabel, MemorySyncStatus, +}; pub use traits::{Memory, MemoryCategory, MemoryEntry, NamespaceSummary, RecallOpts}; pub use tree::{ all_memory_tree_controller_schemas, all_memory_tree_registered_controllers, diff --git a/src/openhuman/memory/sync_status/mod.rs b/src/openhuman/memory/sync_status/mod.rs new file mode 100644 index 000000000..38f7a642d --- /dev/null +++ b/src/openhuman/memory/sync_status/mod.rs @@ -0,0 +1,24 @@ +//! Memory sync status surface (#1136 — simplified rewrite). +//! +//! The earlier push-based design (phase events from each provider's +//! sync loop, persisted KV store, subscriber that mirrored events +//! into storage) was replaced because it drifted from reality — +//! "downloading 0/0" was a common lie while the chunks table told +//! the truth. The pull-based replacement is one SQL query against +//! `mem_tree_chunks` GROUPED BY `source_kind` on each RPC call. +//! +//! Public surface: +//! +//! * [`MemorySyncStatus`] / [`FreshnessLabel`] — what the RPC returns +//! * `openhuman.memory_sync_status_list` — handler in [`rpc`] +//! * Controller registration via [`schemas::all_registered_controllers`] + +pub mod rpc; +pub mod schemas; +pub mod types; + +pub use schemas::{ + all_controller_schemas as all_memory_sync_status_controller_schemas, + all_registered_controllers as all_memory_sync_status_registered_controllers, +}; +pub use types::{FreshnessLabel, MemorySyncStatus}; diff --git a/src/openhuman/memory/sync_status/rpc.rs b/src/openhuman/memory/sync_status/rpc.rs new file mode 100644 index 000000000..44e109ca4 --- /dev/null +++ b/src/openhuman/memory/sync_status/rpc.rs @@ -0,0 +1,142 @@ +//! JSON-RPC handler for `openhuman.memory_sync_status_list` (#1136). +//! +//! Single SQL query against `mem_tree_chunks`. Two layers of metrics: +//! +//! * **Lifetime** — `chunks_synced` (total ingested), `chunks_pending` +//! (`embedding IS NULL` = still in the extract+embed queue, not +//! yet appended to the source-tree buffer). +//! +//! * **Active sync wave** — `batch_total` / `batch_processed`. The +//! wave is identified by a *time-cluster anchor*: the earliest +//! chunk within `WAVE_WINDOW_MS` of the most recent chunk (per +//! provider). A typical sync ingests its whole batch in seconds, +//! so a 10-minute window cleanly captures one wave; if no new +//! chunks arrive, the anchor stays put. Two syncs <10min apart +//! merge into one wave (acceptable — they're contiguous activity). +//! +//! Stateless: no per-process Mutex, no persisted side table. Pure SQL +//! + the chunks table. Survives restart, safe across multiple core +//! processes. +//! +//! Trade-off: pending chunks older than `WAVE_WINDOW_MS` (e.g., +//! leftovers from a stuck earlier wave when the worker was offline) +//! show up in lifetime `chunks_pending` but not in `batch_total` — +//! deliberately, since they shouldn't pollute the active wave's +//! progress signal. + +use crate::openhuman::config::Config; +use crate::openhuman::memory::tree::store::with_connection; +use crate::rpc::RpcOutcome; + +use super::types::{FreshnessLabel, MemorySyncStatus, StatusListResponse}; + +/// Sliding window used to identify a "current sync wave". Chunks +/// within this many ms of `MAX(created_at_ms)` for a provider count +/// as part of the wave; older chunks fall out. +const WAVE_WINDOW_MS: i64 = 10 * 60 * 1000; + +/// `openhuman.memory_sync_status_list` — one row per provider that +/// has chunks, with lifetime + active-wave counters and a freshness +/// label. +pub async fn status_list_rpc(config: &Config) -> Result, String> { + tracing::debug!("[memory_sync_status][rpc] status_list"); + + let config = config.clone(); + let statuses: Vec = tokio::task::spawn_blocking(move || { + with_connection(&config, |conn| -> anyhow::Result> { + // Provider parsed from `source_id` prefix (substring before + // first ':'); falls back to `source_kind` when no prefix. + // + // `provider_chunks` projects per-row provider + the columns + // we need. `provider_pending` flags providers that still + // have at least one chunk waiting for an embedding — + // `wave_anchors` is gated on this so a fully-drained + // provider gets `batch_total = batch_processed = 0` (the + // UI then hides the progress bar instead of rendering a + // completed one for an idle connection). `wave_anchors` + // finds the earliest chunk within WAVE_WINDOW_MS of the + // most recent — the wave's start. The outer SELECT joins + // back to count both lifetime and in-wave totals. + let mut stmt = conn.prepare( + "WITH provider_chunks AS ( \ + SELECT \ + CASE \ + WHEN INSTR(source_id, ':') > 0 \ + THEN SUBSTR(source_id, 1, INSTR(source_id, ':') - 1) \ + ELSE source_kind \ + END AS provider, \ + created_at_ms, \ + embedding, \ + timestamp_ms \ + FROM mem_tree_chunks \ + ), \ + provider_max AS ( \ + SELECT provider, MAX(created_at_ms) AS max_created \ + FROM provider_chunks \ + GROUP BY provider \ + ), \ + provider_pending AS ( \ + SELECT provider, \ + SUM(CASE WHEN embedding IS NULL THEN 1 ELSE 0 END) AS pending \ + FROM provider_chunks \ + GROUP BY provider \ + ), \ + wave_anchors AS ( \ + SELECT p.provider, MIN(p.created_at_ms) AS anchor \ + FROM provider_chunks p \ + JOIN provider_max m ON p.provider = m.provider \ + JOIN provider_pending pp ON p.provider = pp.provider \ + WHERE pp.pending > 0 \ + AND p.created_at_ms >= m.max_created - ?1 \ + GROUP BY p.provider \ + ) \ + SELECT \ + p.provider, \ + COUNT(*) AS chunks_synced, \ + SUM(CASE WHEN p.embedding IS NULL THEN 1 ELSE 0 END) AS chunks_pending, \ + SUM(CASE WHEN w.anchor IS NOT NULL \ + AND p.created_at_ms >= w.anchor \ + THEN 1 ELSE 0 END) AS batch_total, \ + SUM(CASE WHEN w.anchor IS NOT NULL \ + AND p.created_at_ms >= w.anchor \ + AND p.embedding IS NOT NULL \ + THEN 1 ELSE 0 END) AS batch_processed, \ + MAX(p.timestamp_ms) AS last_chunk_at_ms \ + FROM provider_chunks p \ + LEFT JOIN wave_anchors w ON p.provider = w.provider \ + GROUP BY p.provider \ + ORDER BY last_chunk_at_ms DESC", + )?; + let now_ms = chrono::Utc::now().timestamp_millis(); + let iter = stmt.query_map([WAVE_WINDOW_MS], |row| { + let provider: String = row.get(0)?; + let chunks_synced: i64 = row.get(1)?; + let chunks_pending: i64 = row.get(2)?; + let batch_total: i64 = row.get(3)?; + let batch_processed: i64 = row.get(4)?; + let last_chunk_at_ms: Option = row.get(5)?; + Ok(MemorySyncStatus { + provider, + chunks_synced: chunks_synced.max(0) as u64, + chunks_pending: chunks_pending.max(0) as u64, + batch_total: batch_total.max(0) as u64, + batch_processed: batch_processed.max(0) as u64, + last_chunk_at_ms, + freshness: FreshnessLabel::from_age_ms(last_chunk_at_ms, now_ms), + }) + })?; + let out = iter.collect::, _>>()?; + Ok(out) + }) + }) + .await + .map_err(|e| format!("spawn_blocking join failed: {e}"))? + .map_err(|e| format!("memory_tree DB access failed: {e:#}"))?; + + let log = format!( + "[memory_sync_status][rpc] status_list returning {} row(s)", + statuses.len() + ); + tracing::debug!("{log}"); + Ok(RpcOutcome::single_log(StatusListResponse { statuses }, log)) +} diff --git a/src/openhuman/memory/sync_status/schemas.rs b/src/openhuman/memory/sync_status/schemas.rs new file mode 100644 index 000000000..c0a4c1733 --- /dev/null +++ b/src/openhuman/memory/sync_status/schemas.rs @@ -0,0 +1,84 @@ +//! Controller-registry schemas for `openhuman.memory_sync_status_list`. +//! +//! Wired into `src/core/all.rs` via the `all_memory_sync_status_*` +//! re-exports in `super::mod`. Single method now — see `rpc.rs` and +//! `types.rs` for the simplified design (#1136 rewrite). + +use serde_json::{Map, Value}; + +use crate::core::all::{ControllerFuture, RegisteredController}; +use crate::core::{ControllerSchema, FieldSchema, TypeSchema}; +use crate::openhuman::config::ops::load_config_with_timeout; +use crate::rpc::RpcOutcome; + +use super::rpc; + +pub fn all_controller_schemas() -> Vec { + vec![schemas("status_list")] +} + +pub fn all_registered_controllers() -> Vec { + vec![RegisteredController { + schema: schemas("status_list"), + handler: handle_status_list, + }] +} + +pub fn schemas(function: &str) -> ControllerSchema { + match function { + "status_list" => ControllerSchema { + namespace: "memory_sync", + function: "status_list", + description: + "List one row per data-source kind that has chunks in the memory tree. Counts \ + are pulled live from `mem_tree_chunks` so the snapshot is always exact.", + inputs: vec![], + outputs: vec![FieldSchema { + name: "statuses", + ty: TypeSchema::Array(Box::new(TypeSchema::Ref("MemorySyncStatus"))), + comment: "One row per `source_kind` with chunk count + freshness label.", + required: true, + }], + }, + other => panic!("unknown memory_sync schema function: {other}"), + } +} + +fn handle_status_list(_params: Map) -> ControllerFuture { + Box::pin(async move { + let config = load_config_with_timeout().await?; + to_json(rpc::status_list_rpc(&config).await?) + }) +} + +fn to_json(outcome: RpcOutcome) -> Result { + outcome.into_cli_compatible_json() +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn registers_only_status_list() { + let regs = all_registered_controllers(); + assert_eq!(regs.len(), 1); + assert_eq!(regs[0].schema.function, "status_list"); + } + + #[test] + fn schema_status_list_has_no_inputs_and_one_output() { + let s = schemas("status_list"); + assert_eq!(s.namespace, "memory_sync"); + assert_eq!(s.function, "status_list"); + assert!(s.inputs.is_empty()); + assert_eq!(s.outputs.len(), 1); + assert_eq!(s.outputs[0].name, "statuses"); + } + + #[test] + #[should_panic(expected = "unknown memory_sync schema function")] + fn schemas_panics_on_unknown_function() { + schemas("nope"); + } +} diff --git a/src/openhuman/memory/sync_status/types.rs b/src/openhuman/memory/sync_status/types.rs new file mode 100644 index 000000000..d0f88f41a --- /dev/null +++ b/src/openhuman/memory/sync_status/types.rs @@ -0,0 +1,126 @@ +//! Memory sync status — types (#1136, simplified rewrite). +//! +//! The original implementation tracked phase + counters via push-based +//! events from each provider's sync loop. That was racy, lied about +//! "downloading 0/0" while work was in flight, and required maintaining +//! a parallel KV store. Replaced with a pull model: count chunks in +//! `mem_tree_chunks` GROUPED BY source_kind on each RPC. The chunks +//! table is the source of truth — if a chunk exists, that source has +//! synced something; the count is exact at any moment. +//! +//! Activity-freshness is derived from `MAX(timestamp_ms)` per group. + +use serde::Serialize; + +/// User-facing label derived from how recently chunks were ingested. +/// +/// Computed at RPC time, not stored. Boundaries are deliberate: +/// `Active` matches "currently syncing" (a fresh chunk in the last 30s +/// suggests live ingest), `Recent` covers "synced this session", and +/// `Idle` is everything older. +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum FreshnessLabel { + Active, + Recent, + Idle, +} + +impl FreshnessLabel { + /// Map `last_chunk_at_ms` to a label using `now_ms` as reference. + /// Returns `Idle` when `last_chunk_at_ms` is `None`. + pub fn from_age_ms(last_chunk_at_ms: Option, now_ms: i64) -> Self { + match last_chunk_at_ms { + None => Self::Idle, + Some(ts) => { + let age = now_ms.saturating_sub(ts); + if age <= 30_000 { + Self::Active + } else if age <= 5 * 60_000 { + Self::Recent + } else { + Self::Idle + } + } + } + } +} + +/// One row per provider (slack/gmail/discord/notion/…) that has +/// produced chunks. The provider name is parsed from each chunk's +/// `source_id` prefix (everything before the first `:`). +#[derive(Clone, Debug, Serialize)] +pub struct MemorySyncStatus { + /// Specific provider — `"slack"`, `"gmail"`, `"discord"`, + /// `"telegram"`, `"whatsapp"`, `"notion"`, `"meeting_notes"`, + /// `"drive_docs"`, etc. Derived from `source_id` prefix; falls + /// back to the broad `source_kind` category for chunks whose + /// `source_id` has no `:` separator. + pub provider: String, + /// Total chunks in `mem_tree_chunks` for this source_kind. + pub chunks_synced: u64, + /// Chunks fetched + stored but not yet processed by the extract+embed + /// background worker (`embedding IS NULL`). Lifetime metric — counts + /// every still-pending chunk regardless of when it was ingested. + pub chunks_pending: u64, + /// Total chunks in the *current sync wave* — i.e., chunks created + /// at-or-after the oldest currently-pending chunk's `created_at_ms`. + /// When `chunks_pending == 0` this is also 0 (no active wave). + pub batch_total: u64, + /// Of `batch_total`, how many have been processed (`embedding IS NOT + /// NULL`) since the wave started. Progress fill = `batch_processed / + /// batch_total`. + pub batch_processed: u64, + /// Most recent chunk's `timestamp_ms` for this source_kind, or + /// `None` if no chunks yet. + pub last_chunk_at_ms: Option, + /// Derived from `last_chunk_at_ms` at RPC time. + pub freshness: FreshnessLabel, +} + +/// Wire shape of `openhuman.memory_sync_status_list`. +#[derive(Clone, Debug, Serialize)] +pub struct StatusListResponse { + pub statuses: Vec, +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn freshness_label_active_within_30s() { + let now = 1_777_000_000_000; + assert_eq!( + FreshnessLabel::from_age_ms(Some(now - 1_000), now), + FreshnessLabel::Active + ); + assert_eq!( + FreshnessLabel::from_age_ms(Some(now - 29_999), now), + FreshnessLabel::Active + ); + } + + #[test] + fn freshness_label_recent_between_30s_and_5min() { + let now = 1_777_000_000_000; + assert_eq!( + FreshnessLabel::from_age_ms(Some(now - 30_001), now), + FreshnessLabel::Recent + ); + assert_eq!( + FreshnessLabel::from_age_ms(Some(now - 4 * 60_000), now), + FreshnessLabel::Recent + ); + } + + #[test] + fn freshness_label_idle_beyond_5min() { + let now = 1_777_000_000_000; + assert_eq!( + FreshnessLabel::from_age_ms(Some(now - 5 * 60_000 - 1), now), + FreshnessLabel::Idle + ); + assert_eq!(FreshnessLabel::from_age_ms(None, now), FreshnessLabel::Idle); + } +}