feat(memory): Notion doc-aware versioned memory tree + page-content ingest

app/src/components/intelligence/MemoryWorkspace.test.tsx

@@ -0,0 +1,69 @@
+import { act, fireEvent, render, screen, waitFor } from '@testing-library/react';
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
+
+import { memoryTreeGraphExport } from '../../utils/tauriCommands';
+import { MemoryWorkspace } from './MemoryWorkspace';
+
+// Stub the i18n hook and the heavy child panels so the test renders just the
+// MemoryWorkspace shell (graph fetch effect + the refresh control + poll).
+vi.mock('../../lib/i18n/I18nContext', () => ({ useT: () => ({ t: (k: string) => k }) }));
+vi.mock('./MemoryGraph', () => ({ MemoryGraph: () => null }));
+vi.mock('./MemorySourcesRegistry', () => ({ MemorySourcesRegistry: () => null }));
+vi.mock('./MemoryTreeStatusPanel', () => ({ MemoryTreeStatusPanel: () => null }));
+vi.mock('./ObsidianVaultSection', () => ({ ObsidianVaultSection: () => null }));
+vi.mock('./SyncAuditPanel', () => ({ SyncAuditPanel: () => null }));
+vi.mock('./WhatsAppMemorySection', () => ({ WhatsAppMemorySection: () => null }));
+vi.mock('../../utils/tauriCommands', () => ({
+  memoryTreeGraphExport: vi.fn().mockResolvedValue({ nodes: [], edges: [] }),
+  memoryTreeFlushNow: vi.fn().mockResolvedValue(undefined),
+  memoryTreeResetTree: vi.fn().mockResolvedValue(undefined),
+  memoryTreeWipeAll: vi.fn().mockResolvedValue(undefined),
+}));
+
+const graphExportMock = vi.mocked(memoryTreeGraphExport);
+
+describe('<MemoryWorkspace />', () => {
+  beforeEach(() => {
+    graphExportMock.mockClear();
+  });
+  afterEach(() => {
+    vi.useRealTimers();
+  });
+
+  it('re-exports the graph when the refresh button is clicked', async () => {
+    render(<MemoryWorkspace />);
+    // Mount runs the initial graph load.
+    await waitFor(() => expect(graphExportMock).toHaveBeenCalledTimes(1));
+
+    fireEvent.click(screen.getByTestId('memory-graph-refresh'));
+
+    // Bumping graphVersion re-runs the load effect.
+    await waitFor(() => expect(graphExportMock).toHaveBeenCalledTimes(2));
+  });
+
+  it('polls the graph on a 30s tick, and skips the tick while the tab is hidden', async () => {
+    vi.useFakeTimers();
+    render(<MemoryWorkspace />);
+    // Flush the mount-time async load under fake timers.
+    await act(async () => {
+      await vi.advanceTimersByTimeAsync(0);
+    });
+    const afterMount = graphExportMock.mock.calls.length;
+
+    // Visible tab → the 30s tick re-pulls the graph.
+    await act(async () => {
+      await vi.advanceTimersByTimeAsync(30_000);
+    });
+    expect(graphExportMock.mock.calls.length).toBeGreaterThan(afterMount);
+
+    // Backgrounded tab → the tick is skipped (no extra RPC churn).
+    const hiddenSpy = vi.spyOn(document, 'hidden', 'get').mockReturnValue(true);
+    const afterVisible = graphExportMock.mock.calls.length;
+    await act(async () => {
+      await vi.advanceTimersByTimeAsync(30_000);
+    });
+    expect(graphExportMock.mock.calls.length).toBe(afterVisible);
+
+    hiddenSpy.mockRestore();
+  });
+});

app/src/components/intelligence/MemoryWorkspace.tsx

@@ -111,6 +111,22 @@ export function MemoryWorkspace({ onToast }: MemoryWorkspaceProps) {
     };
   }, []);
 
+  // Live refresh: re-pull the graph every 30s while this tab is mounted so it
+  // reflects background tree growth (e.g. seal_document jobs draining as
+  // Notion syncs) without a manual refresh. The Memory tab unmounts this
+  // component when inactive, which clears the interval — so the poll only runs
+  // while the tab is actually open. Ticks are skipped while the window is
+  // backgrounded to avoid needless RPC churn; the next visible tick catches up.
+  useEffect(() => {
+    const GRAPH_POLL_MS = 30_000;
+    const id = setInterval(() => {
+      if (typeof document !== 'undefined' && document.hidden) return;
+      console.debug('[ui-flow][memory-workspace] graph poll tick → bump version');
+      setGraphVersion(v => v + 1);
+    }, GRAPH_POLL_MS);
+    return () => clearInterval(id);
+  }, []);
+
   const handleWipe = useCallback(async () => {
     // Two-step confirm so accidental clicks can't nuke a workspace.
     const ok = window.confirm(t('workspace.wipeConfirm'));
@@ -236,6 +252,17 @@ export function MemoryWorkspace({ onToast }: MemoryWorkspaceProps) {
         data-testid="memory-actions">
         <ModeToggle mode={mode} onChange={setMode} />
         <div className="flex flex-wrap items-center gap-2">
+          <button
+            type="button"
+            onClick={() => setGraphVersion(v => v + 1)}
+            data-testid="memory-graph-refresh"
+            className="inline-flex items-center gap-2 rounded-lg
+                       border border-stone-200 dark:border-neutral-700 bg-white dark:bg-neutral-900 px-4 py-2 text-sm font-semibold
+                       text-stone-700 dark:text-neutral-200 shadow-sm transition-colors hover:bg-stone-50 dark:hover:bg-neutral-800
+                       focus:outline-none focus:ring-2 focus:ring-stone-200"
+            title={t('common.refresh')}>
+            <RefreshIcon /> {t('common.refresh')}
+          </button>
           <button
             type="button"
             onClick={handleWipe}

app/src/components/intelligence/memoryGraphLayout.test.ts

@@ -41,10 +41,15 @@ describe('memoryGraphLayout', () => {
     expect(nodeColor(contact())).toBe(CONTACT_COLOR);
   });
 
-  it('nodeRadius grows with level and is fixed for chunk/contact', () => {
+  it('nodeRadius grows with level (capped) and is fixed for chunk/contact', () => {
     expect(nodeRadius(summary({ level: 0 }))).toBe(5);
     expect(nodeRadius(summary({ level: 3 }))).toBe(12.5);
-    expect(nodeRadius(summary({ level: 99 }))).toBe(252.5);
+    // Capped at 14: document merge-tier nodes live at a large synthetic level
+    // (MERGE_LEVEL_BASE = 1000+), so the raw `5 + level*2.5` is clamped to keep
+    // the d3 layout/collision sane instead of rendering giant discs.
+    expect(nodeRadius(summary({ level: 4 }))).toBe(14); // 5 + 4*2.5 = 15 → capped
+    expect(nodeRadius(summary({ level: 99 }))).toBe(14);
+    expect(nodeRadius(summary({ level: 1001 }))).toBe(14);
     expect(nodeRadius(contact())).toBe(9);
     expect(nodeRadius(chunk())).toBe(3);
   });

app/src/components/intelligence/memoryGraphLayout.ts

@@ -40,7 +40,14 @@ export const SOURCE_COLOR = '#F97316'; // synthetic source root nodes
 /** Layout is computed in this fixed coordinate space; the renderer pans/zooms it. */
 export const VIEWPORT_W = 1100;
 export const VIEWPORT_H = 640;
-export const ZOOM_MIN = 0.3;
+// Lower bound shared by auto-fit framing and manual wheel zoom-out. Kept very
+// small (20× zoom-out) so large clouds — e.g. a Notion connection's hundreds
+// of page-chunk leaves — can be framed in full. At 0.3 the auto-fit was
+// clamped above the scale needed to show every node, so big graphs rendered
+// "too zoomed in" with the outer nodes spilling off-screen. Using one shared
+// floor (rather than a separate, lower auto-fit floor) avoids a zoom-snap
+// where the first wheel tick would jump back up to the manual floor.
+export const ZOOM_MIN = 0.05;
 export const ZOOM_MAX = 4;
 
 export function levelColor(level: number | null | undefined): string {
@@ -57,7 +64,16 @@ export function nodeColor(node: GraphNode): string {
 
 export function nodeRadius(node: GraphNode): number {
   if (node.kind === 'source') return 16;
-  if (node.kind === 'summary') return 5 + (node.level ?? 0) * 2.5;
+  if (node.kind === 'summary') {
+    // Higher levels render slightly larger, but the size MUST be capped:
+    // document source trees place their cross-document merge tier at a large
+    // synthetic level (MERGE_LEVEL_BASE = 1000+), so the raw `level * 2.5`
+    // would explode to thousands of px — rendering giant discs and, via the
+    // `forceCollide(nodeRadius + 2)` term, blowing the whole layout apart.
+    // The cap keeps merge nodes the largest summaries without distorting it.
+    const level = node.level ?? 0;
+    return Math.min(5 + level * 2.5, 14);
+  }
   if (node.kind === 'contact') return 9;
   return 3; // chunk / document leaf
 }

src/openhuman/composio/ops_tests.rs

@@ -590,6 +590,8 @@ async fn composio_delete_connection_clear_memory_cascades_source_tree_and_conten
                 sealed_at: ts,
                 deleted: false,
                 embedding: None,
+                doc_id: None,
+                version_ms: None,
             },
             None,
             "test/model@3",

src/openhuman/memory/ingest_pipeline.rs

@@ -88,7 +88,7 @@ pub async fn ingest_chat(
             Some(c) => c,
             None => return Ok(IngestResult::empty(source_id)),
         };
-    persist(config, source_id, canonical).await
+    persist(config, source_id, canonical, None).await
 }
 
 /// Ingest an email thread: canonicalise → chunk → fast-score → persist → enqueue
@@ -110,7 +110,7 @@ pub async fn ingest_email(
             Some(c) => c,
             None => return Ok(IngestResult::empty(source_id)),
         };
-    persist(config, source_id, canonical).await
+    persist(config, source_id, canonical, None).await
 }
 
 /// Ingest a single document: canonicalise → chunk → fast-score → persist →
@@ -133,10 +133,39 @@ pub async fn ingest_document_with_scope(
     doc: DocumentInput,
     path_scope: Option<String>,
 ) -> Result<IngestResult> {
-    if already_ingested(config, SourceKind::Document, source_id).await? {
+    ingest_document_versioned(config, source_id, owner, tags, doc, path_scope, None).await
+}
+
+/// Like [`ingest_document_with_scope`] but version-aware.
+///
+/// When `version_ms` is `Some`, the source-ingest gate is keyed by
+/// `{source_id}@{version_ms}` instead of the bare `source_id`, so a later
+/// revision of the SAME document (same `source_id`) is admitted
+/// **non-destructively** — the prior version's chunks are left in place and
+/// the new version is ingested alongside them. Chunks keep the clean
+/// `source_id` (their `doc_id`), and the retrieval layer surfaces only the
+/// latest version per document.
+///
+/// `version_ms = None` is identical to [`ingest_document_with_scope`]
+/// (bare-`source_id` gate), so non-versioned document sources are unaffected.
+pub async fn ingest_document_versioned(
+    config: &Config,
+    source_id: &str,
+    owner: &str,
+    tags: Vec<String>,
+    doc: DocumentInput,
+    path_scope: Option<String>,
+    version_ms: Option<i64>,
+) -> Result<IngestResult> {
+    let gate_key = match version_ms {
+        Some(v) => format!("{source_id}@{v}"),
+        None => source_id.to_string(),
+    };
+    if already_ingested(config, SourceKind::Document, &gate_key).await? {
         log::debug!(
-            "[memory::ingest_pipeline] skip ingest_document — source_id_hash={} already ingested",
-            redact(source_id)
+            "[memory::ingest_pipeline] skip ingest_document — source_id_hash={} version_ms={:?} already ingested",
+            redact(source_id),
+            version_ms
         );
         return Ok(IngestResult::already_ingested(source_id));
     }
@@ -146,7 +175,7 @@ pub async fn ingest_document_with_scope(
         Some(c) => c,
         None => return Ok(IngestResult::empty(source_id)),
     };
-    persist(config, source_id, canonical).await
+    persist(config, source_id, canonical, version_ms).await
 }
 
 /// Best-effort pre-canonicalisation check. The transactional claim inside
@@ -168,6 +197,7 @@ async fn persist(
     config: &Config,
     source_id: &str,
     canonical: CanonicalisedSource,
+    gate_version_ms: Option<i64>,
 ) -> Result<IngestResult> {
     let source_kind_for_store = canonical.metadata.source_kind;
 
@@ -274,10 +304,22 @@ async fn persist(
             // genuine replays without blocking legitimate appends.
             if source_kind_for_store == SourceKind::Document {
                 let now_ms = chrono::Utc::now().timestamp_millis();
+                // Versioned sources (Notion) claim a per-version gate key
+                // `{source_id}@{version_ms}` so a later revision of the SAME
+                // document is admitted (non-destructively, alongside the
+                // prior version) instead of short-circuiting on the bare
+                // source_id. Chunks themselves keep the clean source_id so
+                // per-document grouping (doc_id) stays stable. Non-versioned
+                // documents (version_ms = None) use the bare source_id as
+                // before — behaviour unchanged.
+                let gate_key = match gate_version_ms {
+                    Some(v) => format!("{source_id_for_store}@{v}"),
+                    None => source_id_for_store.clone(),
+                };
                 let claimed = chunk_store::claim_source_ingest_tx(
                     &tx,
                     source_kind_for_store,
-                    &source_id_for_store,
+                    &gate_key,
                     now_ms,
                 )?;
                 if !claimed {

src/openhuman/memory/read_rpc.rs

@@ -1672,6 +1672,14 @@ pub async fn wipe_all_rpc(config: &Config) -> Result<RpcOutcome<WipeAllResponse>
             "mem_tree_summaries",
             "mem_tree_trees",
             "mem_tree_chunks",
+            // Source-level ingest gate. MUST be cleared on wipe: otherwise the
+            // chunks are gone but `(source_kind, source_id[@version])` stays
+            // claimed, so the next sync sees `already_ingested` and writes 0
+            // chunks / enqueues 0 seal jobs — a wiped source can never
+            // rebuild. (Previously masked for documents by the old
+            // delete-first re-ingest path, which has been removed in favour of
+            // non-destructive versioned ingest.)
+            "mem_tree_ingested_sources",
         ];
         let rows_deleted: u64 = with_connection(&cfg, |conn| {
             let tx = conn.unchecked_transaction()?;

src/openhuman/memory/read_rpc_tests.rs

@@ -947,3 +947,43 @@ async fn vault_health_check_reports_writable_and_obsidian_registered_when_ready(
         outcome.logs[0]
     );
 }
+
+/// Regression: `wipe_all` MUST also clear the source-ingest gate
+/// (`mem_tree_ingested_sources`). Before the fix it cleared chunks/summaries
+/// but left the gate claimed, so a wiped document source could never
+/// re-ingest — the next sync saw `already_ingested` and wrote 0 chunks / 0
+/// seal jobs. This pins that a wipe leaves the gate empty so re-sync works.
+#[tokio::test]
+async fn wipe_all_clears_ingest_gate() {
+    use crate::openhuman::memory_store::chunks::store as chunk_store;
+    use crate::openhuman::memory_store::chunks::types::SourceKind;
+
+    let (_tmp, cfg) = test_config();
+    let gate_key = "notion:conn-1:page-abc@1700000000000";
+
+    // Claim the gate exactly as a document ingest does.
+    chunk_store::with_connection(&cfg, |conn| {
+        let tx = conn.unchecked_transaction()?;
+        let claimed = chunk_store::claim_source_ingest_tx(
+            &tx,
+            SourceKind::Document,
+            gate_key,
+            1_700_000_000_000,
+        )?;
+        assert!(claimed, "first claim should succeed");
+        tx.commit()?;
+        Ok(())
+    })
+    .unwrap();
+    assert!(
+        chunk_store::is_source_ingested(&cfg, SourceKind::Document, gate_key).unwrap(),
+        "gate must be claimed before wipe"
+    );
+
+    wipe_all_rpc(&cfg).await.expect("wipe_all_rpc");
+
+    assert!(
+        !chunk_store::is_source_ingested(&cfg, SourceKind::Document, gate_key).unwrap(),
+        "wipe_all must clear mem_tree_ingested_sources so a wiped source can re-ingest"
+    );
+}