feat(memory): per-source memory sync status (#1136) (#1250)

Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-07-27 21:08:00 +00:00 · 2026-05-05 15:34:28 -07:00
parent 65032870f2
commit fff1339532
11 changed files with 878 additions and 70 deletions
@@ -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<typeof import('../../services/memorySyncService')>(
+    '../../services/memorySyncService'
+  );
+  return { ...actual, memorySyncStatusList: (...args: unknown[]) => mockStatusList(...args) };
+});
+
+function makeStatus(overrides: Partial<MemorySyncStatus> = {}): 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('<MemorySyncConnections />', () => {
+  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(<MemorySyncConnections />);
+    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(<MemorySyncConnections />);
+    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(<MemorySyncConnections />);
+    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(<MemorySyncConnections />);
+    await waitFor(() => {
+      expect(screen.getByText(/Failed to load sync status/)).toBeTruthy();
+      expect(screen.getByText(/rpc unavailable/)).toBeTruthy();
+    });
+  });
+});
@@ -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<FreshnessLabel, string> = {
+  active: 'Active',
+  recent: 'Recent',
+  idle: 'Idle',
+};
+
+const PROVIDER_LABEL: Record<string, string> = {
+  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 (
+    <div
+      className="rounded-lg border border-stone-200 bg-white p-4 shadow-sm"
+      data-testid={`memory-sync-card-${status.provider}`}>
+      <div className="flex items-start justify-between gap-3">
+        <div className="min-w-0 flex-1">
+          <div className="flex items-center gap-2">
+            <span className="font-medium text-stone-900">{label}</span>
+            <span
+              className={`rounded-md px-2 py-0.5 text-xs font-medium ${freshnessBadgeClass(status.freshness)}`}
+              data-testid={`memory-sync-freshness-${status.provider}`}>
+              {FRESHNESS_LABEL[status.freshness]}
+            </span>
+          </div>
+          <div className="mt-1 flex items-center gap-3 text-xs text-stone-500">
+            <span data-testid={`memory-sync-chunks-${status.provider}`}>
+              {lifetime.toLocaleString()} chunks
+            </span>
+            {lastSync && <span>Last chunk {lastSync}</span>}
+          </div>
+        </div>
+      </div>
+
+      {showProgress && (
+        <div className="mt-3" data-testid={`memory-sync-progress-${status.provider}`}>
+          <div className="h-1.5 w-full overflow-hidden rounded-full bg-stone-100">
+            <div
+              className="h-full bg-ocean-400 transition-all"
+              style={{ width: `${pct}%` }}
+              role="progressbar"
+              aria-valuenow={batchProcessed}
+              aria-valuemin={0}
+              aria-valuemax={batchTotal}
+            />
+          </div>
+          <div className="mt-1 text-xs text-stone-500">
+            <span data-testid={`memory-sync-pending-${status.provider}`}>
+              {batchProcessed.toLocaleString()} of {batchTotal.toLocaleString()} processed
+            </span>
+            {pending > 0 && (
+              <span className="text-stone-400"> · {pending.toLocaleString()} pending</span>
+            )}
+          </div>
+        </div>
+      )}
+    </div>
+  );
+}
+
+export function MemorySyncConnections({ pollIntervalMs }: MemorySyncConnectionsProps) {
+  const [statuses, setStatuses] = useState<MemorySyncStatus[]>([]);
+  const [loadError, setLoadError] = useState<string | null>(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 (
+      <section className="memory-sync-connections" data-testid="memory-sync-connections">
+        <h3 className="text-sm font-semibold text-stone-700">Memory sources</h3>
+        <p className="mt-2 text-xs text-stone-500">Loading…</p>
+      </section>
+    );
+  }
+
+  if (loadError) {
+    return (
+      <section className="memory-sync-connections" data-testid="memory-sync-connections">
+        <h3 className="text-sm font-semibold text-stone-700">Memory sources</h3>
+        <p className="mt-2 rounded-md bg-coral-50 p-2 text-xs text-coral-800">
+          Failed to load sync status: {loadError}
+        </p>
+      </section>
+    );
+  }
+
+  if (statuses.length === 0) {
+    return (
+      <section className="memory-sync-connections" data-testid="memory-sync-connections">
+        <h3 className="text-sm font-semibold text-stone-700">Memory sources</h3>
+        <p className="mt-2 text-xs text-stone-500">
+          No content has been synced into memory yet. Connect an integration to start.
+        </p>
+      </section>
+    );
+  }
+
+  return (
+    <section className="memory-sync-connections" data-testid="memory-sync-connections">
+      <h3 className="text-sm font-semibold text-stone-700">Memory sources</h3>
+      <div className="mt-2 space-y-2">
+        {statuses.map(s => (
+          <SourceCard key={s.provider} status={s} />
+        ))}
+      </div>
+    </section>
+  );
+}
@@ -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<ToastNotification, 'id'>) => void;
@@ -214,89 +215,95 @@ export function MemoryWorkspace({ onToast: _onToast }: MemoryWorkspaceProps) {

  if (isEmpty) {
    return (
-      <div className="flex min-h-[40vh] items-center justify-center">
-        <MemoryEmptyPlaceholder />
+      <div className="space-y-4">
+        <MemorySyncConnections pollIntervalMs={5000} />
+        <div className="flex min-h-[40vh] items-center justify-center">
+          <MemoryEmptyPlaceholder />
+        </div>
      </div>
    );
  }

  return (
-    <section
-      className="memory-workspace-root relative flex"
-      style={{ height: 'calc(100vh - 16rem)' }}
-      data-testid="memory-workspace">
-      {/* 2-pane base */}
-      <aside
-        className="w-60 shrink-0 overflow-y-auto border-r border-stone-100 bg-stone-50/60"
-        aria-label="Memory navigator">
-        <MemoryNavigator
-          chunks={allChunks}
-          sources={sources}
-          topPeople={topPeople}
-          topTopics={topTopics}
-          selection={selection}
-          onSelectionChange={handleSelectionChange}
-          searchQuery={searchQuery}
-          onSearchChange={handleSearchChange}
-        />
-      </aside>
+    <div className="space-y-4">
+      <MemorySyncConnections pollIntervalMs={5000} />
+      <section
+        className="memory-workspace-root relative flex"
+        style={{ height: 'calc(100vh - 16rem)' }}
+        data-testid="memory-workspace">
+        {/* 2-pane base */}
+        <aside
+          className="w-60 shrink-0 overflow-y-auto border-r border-stone-100 bg-stone-50/60"
+          aria-label="Memory navigator">
+          <MemoryNavigator
+            chunks={allChunks}
+            sources={sources}
+            topPeople={topPeople}
+            topTopics={topTopics}
+            selection={selection}
+            onSelectionChange={handleSelectionChange}
+            searchQuery={searchQuery}
+            onSearchChange={handleSearchChange}
+          />
+        </aside>

-      <main className="flex-1 overflow-y-auto bg-white" aria-label="Result list">
-        <MemoryResultList
-          chunks={filteredChunks}
-          selectedChunkId={selectedChunkId}
-          onSelectChunk={handleSelectChunk}
-        />
-      </main>
+        <main className="flex-1 overflow-y-auto bg-white" aria-label="Result list">
+          <MemoryResultList
+            chunks={filteredChunks}
+            selectedChunkId={selectedChunkId}
+            onSelectChunk={handleSelectChunk}
+          />
+        </main>

-      {/* Detail overlay — fills the entire workspace card */}
-      {selectedChunk && (
-        <div
-          className="absolute inset-0 z-10 flex flex-col bg-canvas-50/95 backdrop-blur-sm
+        {/* Detail overlay — fills the entire workspace card */}
+        {selectedChunk && (
+          <div
+            className="absolute inset-0 z-10 flex flex-col bg-canvas-50/95 backdrop-blur-sm
                     duration-150 motion-safe:animate-fade-in"
-          role="dialog"
-          aria-modal="true"
-          aria-label="Chunk detail">
-          <header
-            className="sticky top-0 z-10 flex items-center justify-between gap-4
+            role="dialog"
+            aria-modal="true"
+            aria-label="Chunk detail">
+            <header
+              className="sticky top-0 z-10 flex items-center justify-between gap-4
                       border-b border-stone-100 bg-white/90 px-6 py-3 backdrop-blur">
-            <div className="min-w-0 flex-1">
-              <p className="truncate font-mono text-xs text-stone-500">
-                <span className="text-stone-400">{selectedChunk.source_kind}</span>
-                {' · '}
-                {selectedChunk.source_id}
-              </p>
-            </div>
-            <button
-              onClick={handleCloseDetail}
-              aria-label="Close detail (Esc)"
-              title="Close (Esc)"
-              className="rounded-lg p-1.5 text-stone-500 transition-colors
+              <div className="min-w-0 flex-1">
+                <p className="truncate font-mono text-xs text-stone-500">
+                  <span className="text-stone-400">{selectedChunk.source_kind}</span>
+                  {' · '}
+                  {selectedChunk.source_id}
+                </p>
+              </div>
+              <button
+                onClick={handleCloseDetail}
+                aria-label="Close detail (Esc)"
+                title="Close (Esc)"
+                className="rounded-lg p-1.5 text-stone-500 transition-colors
                         hover:bg-stone-100 hover:text-stone-900
                         focus:outline-none focus:ring-2 focus:ring-ocean-200">
-              <svg
-                width="18"
-                height="18"
-                viewBox="0 0 24 24"
-                fill="none"
-                stroke="currentColor"
-                strokeWidth="2"
-                strokeLinecap="round"
-                strokeLinejoin="round"
-                aria-hidden="true">
-                <path d="M18 6L6 18" />
-                <path d="M6 6l12 12" />
-              </svg>
-            </button>
-          </header>
+                <svg
+                  width="18"
+                  height="18"
+                  viewBox="0 0 24 24"
+                  fill="none"
+                  stroke="currentColor"
+                  strokeWidth="2"
+                  strokeLinecap="round"
+                  strokeLinejoin="round"
+                  aria-hidden="true">
+                  <path d="M18 6L6 18" />
+                  <path d="M6 6l12 12" />
+                </svg>
+              </button>
+            </header>

-          <div className="flex-1 overflow-y-auto">
-            <div className="mx-auto max-w-4xl px-8 py-6">
-              <MemoryChunkDetail chunk={selectedChunk} onSelectEntity={handleSelectEntity} />
+            <div className="flex-1 overflow-y-auto">
+              <div className="mx-auto max-w-4xl px-8 py-6">
+                <MemoryChunkDetail chunk={selectedChunk} onSelectEntity={handleSelectEntity} />
+              </div>
            </div>
          </div>
-        </div>
-      )}
-    </section>
+        )}
+      </section>
+    </div>
  );
 }
@@ -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> = {}): 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/);
+  });
+});
@@ -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<T>` 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<MemorySyncStatus[]> {
+  log('memory_sync_status_list: calling core RPC');
+  let resp: StatusListResponse;
+  try {
+    resp = await callCoreRpc<StatusListResponse>({ 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;
+}
@@ -152,6 +152,8 @@ fn build_registered_controllers() -> Vec<RegisteredController> {
    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<ControllerSchema> {
    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/<id>` placeholders (SQLite-backed) to save tokens in prompts, with round-trip rewrite helpers.",
        ),
@@ -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,
@@ -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};
@@ -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<RpcOutcome<StatusListResponse>, String> {
+    tracing::debug!("[memory_sync_status][rpc] status_list");
+
+    let config = config.clone();
+    let statuses: Vec<MemorySyncStatus> = tokio::task::spawn_blocking(move || {
+        with_connection(&config, |conn| -> anyhow::Result<Vec<MemorySyncStatus>> {
+            // 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<i64> = 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::<Result<Vec<_>, _>>()?;
+            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))
+}
@@ -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<ControllerSchema> {
+    vec![schemas("status_list")]
+}
+
+pub fn all_registered_controllers() -> Vec<RegisteredController> {
+    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<String, Value>) -> ControllerFuture {
+    Box::pin(async move {
+        let config = load_config_with_timeout().await?;
+        to_json(rpc::status_list_rpc(&config).await?)
+    })
+}
+
+fn to_json<T: serde::Serialize>(outcome: RpcOutcome<T>) -> Result<Value, String> {
+    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");
+    }
+}
@@ -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<i64>, 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<i64>,
+    /// 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<MemorySyncStatus>,
+}
+
+#[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);
+    }
+}