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

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

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"
+    );
+}

src/openhuman/memory_queue/handlers/mod.rs

@@ -45,6 +45,25 @@ fn derive_tree_scope(source_id: &str) -> String {
     source_id.to_string()
 }
 
+/// Whether a chunk's source uses the per-document rollup + versioning path
+/// (Notion). These chunks are deliberately **not** pushed into the flat L0
+/// buffer by `handle_extract` — their tree is built per document-version by
+/// a `SealDocument` job enqueued at ingest time, which rolls each document's
+/// chunks up to a single doc-root and merges it into the connection tree.
+/// Scoped to Notion for now; other `SourceKind::Document` sources
+/// (GitHub/Linear/ClickUp/vault) keep the existing flat behaviour.
+pub(crate) fn uses_document_subtree(
+    chunk: &crate::openhuman::memory_store::chunks::types::Chunk,
+) -> bool {
+    const DOC_SUBTREE_PREFIX: &str = "notion:";
+    chunk.metadata.source_id.starts_with(DOC_SUBTREE_PREFIX)
+        || chunk
+            .metadata
+            .path_scope
+            .as_deref()
+            .is_some_and(|s| s.starts_with(DOC_SUBTREE_PREFIX))
+}
+
 fn emit_build_progress(
     phase: &str,
     step: &str,
@@ -78,9 +97,76 @@ pub async fn handle_job(config: &Config, job: &Job) -> Result<JobOutcome> {
         JobKind::Seal => handle_seal(config, job).await,
         JobKind::FlushStale => handle_flush_stale(config, job).await,
         JobKind::ReembedBackfill => handle_reembed_backfill(config, job).await,
+        JobKind::SealDocument => handle_seal_document(config, job).await,
     }
 }
 
+/// Build (or re-build for a new version) one document's per-doc subtree and
+/// merge its doc-root into the connection tree. See
+/// [`crate::openhuman::memory_tree::tree::seal_document_subtree`].
+async fn handle_seal_document(config: &Config, job: &Job) -> Result<JobOutcome> {
+    use crate::openhuman::memory::util::redact::redact;
+    use crate::openhuman::memory_tree::tree::seal_document_subtree;
+
+    let payload: crate::openhuman::memory_queue::types::SealDocumentPayload =
+        serde_json::from_str(&job.payload_json).context("parse SealDocument payload")?;
+
+    // doc_id (notion:{conn}:{page}) and tree_scope (notion:{conn}) are
+    // recoverable source identifiers — redact them in all logs / error chains.
+    if payload.chunk_ids.is_empty() {
+        log::debug!(
+            "[memory::jobs] seal_document: empty chunk set doc_id={} — nothing to seal",
+            redact(&payload.doc_id)
+        );
+        return Ok(JobOutcome::Done);
+    }
+
+    // One physical tree per connection scope (e.g. notion:{connection_id}).
+    let tree = get_or_create_source_tree(config, &payload.tree_scope)?;
+    let strategy = TreeFactory::from_tree(&tree).label_strategy(config);
+
+    emit_build_progress(
+        "seal_document",
+        "started",
+        Some(&tree.scope),
+        None,
+        Some(payload.chunk_ids.len() as u32),
+        Some(format!(
+            "doc {} v={:?} ({} chunks)",
+            redact(&payload.doc_id),
+            payload.version_ms,
+            payload.chunk_ids.len()
+        )),
+    );
+
+    let doc_root_id = seal_document_subtree(
+        config,
+        &tree,
+        &payload.doc_id,
+        payload.version_ms,
+        &payload.chunk_ids,
+        &strategy,
+    )
+    .await
+    .with_context(|| {
+        format!(
+            "seal_document_subtree failed tree_scope={} doc_id={}",
+            redact(&payload.tree_scope),
+            redact(&payload.doc_id)
+        )
+    })?;
+
+    log::info!(
+        "[memory::jobs] seal_document done tree_scope={} doc_id={} version_ms={:?} doc_root_id={}",
+        redact(&payload.tree_scope),
+        redact(&payload.doc_id),
+        payload.version_ms,
+        doc_root_id
+    );
+    super::worker::wake_workers();
+    Ok(JobOutcome::Done)
+}
+
 async fn handle_extract(config: &Config, job: &Job) -> Result<JobOutcome> {
     let payload: ExtractChunkPayload =
         serde_json::from_str(&job.payload_json).context("parse ExtractChunk payload")?;
@@ -176,7 +262,11 @@ async fn handle_extract(config: &Config, job: &Job) -> Result<JobOutcome> {
     // enqueued inside the SAME transaction that commits the lifecycle update,
     // so a crash anywhere rolls everything back together and prevents the
     // "lifecycle committed but job lost" crash window.
-    let source_job = if result.kept {
+    // Per-document-versioned sources (Notion) skip the flat L0 buffer: their
+    // tree is built by a `SealDocument` job enqueued at ingest, not chunk by
+    // chunk here. We still score + embed the chunk above so chunk-level
+    // semantic search and the entity index stay populated.
+    let source_job = if result.kept && !uses_document_subtree(&chunk) {
         Some(NewJob::append_buffer(&AppendBufferPayload {
             node: NodeRef::Leaf {
                 chunk_id: chunk.id.clone(),

src/openhuman/memory_queue/types.rs

@@ -23,6 +23,11 @@ pub enum JobKind {
     /// vector at the active embedding signature (post model-switch, or the
     /// §7 dim-mismatch slice), then self-continue until none remain.
     ReembedBackfill,
+    /// Build one document version's per-doc subtree (Notion) and merge its
+    /// doc-root into the connection tree. Replaces the per-chunk
+    /// extract→append_buffer tree path for document sources that opt into
+    /// per-document rollup + versioning.
+    SealDocument,
 }
 
 impl JobKind {
@@ -34,6 +39,7 @@ impl JobKind {
             JobKind::Seal => "seal",
             JobKind::FlushStale => "flush_stale",
             JobKind::ReembedBackfill => "reembed_backfill",
+            JobKind::SealDocument => "seal_document",
         }
     }
 
@@ -54,6 +60,7 @@ impl JobKind {
                 ))
             }
             "reembed_backfill" => JobKind::ReembedBackfill,
+            "seal_document" => JobKind::SealDocument,
             other => return Err(anyhow!("unknown JobKind '{other}'")),
         })
     }
@@ -63,7 +70,10 @@ impl JobKind {
     pub fn is_llm_bound(&self) -> bool {
         matches!(
             self,
-            JobKind::ExtractChunk | JobKind::Seal | JobKind::ReembedBackfill
+            JobKind::ExtractChunk
+                | JobKind::Seal
+                | JobKind::ReembedBackfill
+                | JobKind::SealDocument
         )
     }
 }
@@ -269,6 +279,37 @@ impl ReembedBackfillPayload {
     }
 }
 
+/// Build (or re-build for a new version) one document's per-doc subtree and
+/// merge its doc-root into the connection tree. Carries the full leaf chunk
+/// set for the version so the seal handler can run the per-document
+/// side-cascade in one shot (see
+/// [`crate::openhuman::memory_tree::tree::seal_document_subtree`]).
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub struct SealDocumentPayload {
+    /// Connection-level source-tree scope, e.g. `notion:{connection_id}`.
+    /// All of a connection's documents share this one tree.
+    pub tree_scope: String,
+    /// Document identity (the chunk `source_id`, e.g.
+    /// `notion:{connection_id}:{page_id}`).
+    pub doc_id: String,
+    /// Document version as epoch-ms (`last_edited_time`). `None` for sources
+    /// that don't carry a version.
+    pub version_ms: Option<i64>,
+    /// Leaf chunk ids for this version, in document order.
+    pub chunk_ids: Vec<String>,
+}
+
+impl SealDocumentPayload {
+    /// Stable dedupe key — one in-flight seal per (doc, version). A new
+    /// version gets a distinct key so it isn't suppressed by an older one.
+    pub fn dedupe_key(&self) -> String {
+        match self.version_ms {
+            Some(v) => format!("seal_doc:{}@{}", self.doc_id, v),
+            None => format!("seal_doc:{}", self.doc_id),
+        }
+    }
+}
+
 /// One row in `mem_tree_jobs`. `payload_json` is left as a raw string so
 /// callers parse it lazily based on `kind`.
 #[derive(Clone, Debug)]
@@ -367,6 +408,17 @@ impl NewJob {
             max_attempts: None,
         })
     }
+
+    /// Build a [`JobKind::SealDocument`] enqueue request.
+    pub fn seal_document(p: &SealDocumentPayload) -> Result<Self> {
+        Ok(Self {
+            kind: JobKind::SealDocument,
+            payload_json: serde_json::to_string(p)?,
+            dedupe_key: Some(p.dedupe_key()),
+            available_at_ms: None,
+            max_attempts: None,
+        })
+    }
 }
 
 #[cfg(test)]
@@ -381,6 +433,7 @@ mod tests {
             JobKind::Seal,
             JobKind::FlushStale,
             JobKind::ReembedBackfill,
+            JobKind::SealDocument,
         ] {
             assert_eq!(JobKind::parse(k.as_str()).unwrap(), k);
         }
@@ -389,6 +442,45 @@ mod tests {
         assert!(JobKind::parse("digest_daily").is_err());
     }
 
+    #[test]
+    fn seal_document_dedupe_key_is_per_version() {
+        let v1 = SealDocumentPayload {
+            tree_scope: "notion:conn1".into(),
+            doc_id: "notion:conn1:pageA".into(),
+            version_ms: Some(1717000000000),
+            chunk_ids: vec!["c0".into()],
+        };
+        let v2 = SealDocumentPayload {
+            version_ms: Some(1717500000000),
+            ..v1.clone()
+        };
+        // Distinct versions of the same doc get distinct keys, so a newer
+        // revision is never suppressed by an in-flight older one.
+        assert_ne!(v1.dedupe_key(), v2.dedupe_key());
+        assert_eq!(v1.dedupe_key(), "seal_doc:notion:conn1:pageA@1717000000000");
+        // Unversioned falls back to the bare doc id.
+        let unversioned = SealDocumentPayload {
+            version_ms: None,
+            ..v1.clone()
+        };
+        assert_eq!(unversioned.dedupe_key(), "seal_doc:notion:conn1:pageA");
+    }
+
+    #[test]
+    fn seal_document_roundtrips_through_newjob() {
+        let p = SealDocumentPayload {
+            tree_scope: "notion:conn1".into(),
+            doc_id: "notion:conn1:pageA".into(),
+            version_ms: Some(42),
+            chunk_ids: vec!["c0".into(), "c1".into()],
+        };
+        let job = NewJob::seal_document(&p).unwrap();
+        assert_eq!(job.kind, JobKind::SealDocument);
+        let back: SealDocumentPayload = serde_json::from_str(&job.payload_json).unwrap();
+        assert_eq!(back.chunk_ids, vec!["c0".to_string(), "c1".to_string()]);
+        assert_eq!(back.version_ms, Some(42));
+    }
+
     #[test]
     fn job_status_terminality() {
         assert!(!JobStatus::Ready.is_terminal());

src/openhuman/memory_store/chunks/connection.rs

@@ -323,6 +323,17 @@ fn apply_schema(conn: &Connection) -> Result<()> {
     // Phase MD-content (summaries).
     add_column_if_missing(conn, "mem_tree_summaries", "content_path", "TEXT")?;
     add_column_if_missing(conn, "mem_tree_summaries", "content_sha256", "TEXT")?;
+    // Document source-tree versioning: per-doc subtree nodes (Notion etc.)
+    // carry the document identity + version they sealed for, so retrieval can
+    // keep `max(version_ms)` per `doc_id` at read time (latest-wins) without
+    // ever rewriting older subtrees. NULL on merge-tier and chat/email nodes.
+    add_column_if_missing(conn, "mem_tree_summaries", "doc_id", "TEXT")?;
+    add_column_if_missing(conn, "mem_tree_summaries", "version_ms", "INTEGER")?;
+    conn.execute_batch(
+        "CREATE INDEX IF NOT EXISTS idx_mem_tree_summaries_doc_version \
+         ON mem_tree_summaries(tree_id, doc_id, version_ms);",
+    )
+    .context("Failed to create mem_tree_summaries doc/version index")?;
     // Raw-archive pointer column.
     add_column_if_missing(conn, "mem_tree_chunks", "raw_refs_json", "TEXT")?;
     // #1365: is_user flag on indexed entity rows.

src/openhuman/memory_store/chunks/store_tests.rs

@@ -370,6 +370,8 @@ async fn clear_memory_delete_cascades_orphaned_source_tree_and_settles_queued_jo
         sealed_at: ts,
         deleted: false,
         embedding: None,
+        doc_id: None,
+        version_ms: None,
     };
 
     with_connection(&cfg, |conn| {
@@ -587,6 +589,8 @@ fn clear_memory_delete_removes_orphaned_summary_content_file() {
                 sealed_at: ts,
                 deleted: false,
                 embedding: None,
+                doc_id: None,
+                version_ms: None,
             },
             None,
             "test/model@3",

src/openhuman/memory_store/content/atomic.rs

@@ -12,7 +12,7 @@ use std::io::Write;
 use std::path::Path;
 
 use super::compose::{compose_summary_md, split_front_matter, SummaryComposeInput};
-use super::paths::{summary_abs_path, summary_rel_path};
+use super::paths::{summary_rel_path_with_layout, SummaryDiskLayout};
 
 /// Write `bytes` atomically to `abs_path` if the file does not already exist.
 ///
@@ -124,14 +124,37 @@ pub fn stage_summary(
     input: &SummaryComposeInput<'_>,
     scope_slug: &str,
 ) -> anyhow::Result<StagedSummary> {
-    let rel_path = summary_rel_path(input.tree_kind, scope_slug, input.level, input.summary_id);
-    let abs_path = summary_abs_path(
-        content_root,
+    stage_summary_with_layout(content_root, input, scope_slug, SummaryDiskLayout::Standard)
+}
+
+/// Layout-aware variant of [`stage_summary`]. Document source trees pass a
+/// [`SummaryDiskLayout::DocSubtree`] (per-document, versioned) or
+/// [`SummaryDiskLayout::Merge`] (cross-document merge tier) so the on-disk
+/// vault mirrors the logical tree (`notion` → `docs/<page>/v-<ms>` →
+/// `merge`). All other callers use [`stage_summary`] (`Standard`) unchanged.
+pub fn stage_summary_with_layout(
+    content_root: &Path,
+    input: &SummaryComposeInput<'_>,
+    scope_slug: &str,
+    layout: SummaryDiskLayout<'_>,
+) -> anyhow::Result<StagedSummary> {
+    let rel_path = summary_rel_path_with_layout(
         input.tree_kind,
         scope_slug,
         input.level,
         input.summary_id,
+        layout,
     );
+    // Derive the absolute path by joining the relative path components onto
+    // the content root (same join `summary_abs_path` does internally) so the
+    // two stay consistent regardless of layout.
+    let abs_path = {
+        let mut abs = content_root.to_path_buf();
+        for component in rel_path.split('/') {
+            abs.push(component);
+        }
+        abs
+    };
 
     let composed = compose_summary_md(input);
     let body_bytes = composed.body.as_bytes();

src/openhuman/memory_store/content/paths.rs

@@ -104,6 +104,64 @@ pub fn summary_rel_path(
     }
 }
 
+/// On-disk placement for a summary node within a document **source** tree.
+///
+/// Document source trees (Notion) keep one folder per connection but nest
+/// per-document subtrees and the cross-document merge tier beneath it, so the
+/// vault mirrors the logical shape: `notion` → `docs/<page>/v-<ms>` →
+/// `merge`. Non-document trees (chat/email) and the `Standard` variant use
+/// the flat `source-<scope>/L<level>/` layout unchanged.
+#[derive(Clone, Copy, Debug)]
+pub enum SummaryDiskLayout<'a> {
+    /// Flat layout — `source-<scope>/L<level>/…` (chat, email, legacy).
+    Standard,
+    /// A node inside one document's versioned subtree —
+    /// `source-<scope>/docs/<doc_slug>/v-<version_ms>/L<level>/…`.
+    DocSubtree {
+        doc_slug: &'a str,
+        version_ms: Option<i64>,
+    },
+    /// A cross-document merge-tier node — `source-<scope>/merge/L<level>/…`.
+    Merge,
+}
+
+/// Layout-aware variant of [`summary_rel_path`]. For document source trees it
+/// routes per-doc and merge nodes into nested folders; for everything else
+/// (and [`SummaryDiskLayout::Standard`]) it is identical to
+/// [`summary_rel_path`].
+pub fn summary_rel_path_with_layout(
+    tree_kind: SummaryTreeKind,
+    scope_slug: &str,
+    level: u32,
+    summary_id: &str,
+    layout: SummaryDiskLayout<'_>,
+) -> String {
+    match (tree_kind, layout) {
+        (
+            SummaryTreeKind::Source,
+            SummaryDiskLayout::DocSubtree {
+                doc_slug,
+                version_ms,
+            },
+        ) => {
+            let filename = summary_filename(summary_id);
+            let vfolder = match version_ms {
+                Some(v) => format!("v-{v}"),
+                None => "v-unversioned".to_string(),
+            };
+            format!(
+                "{WIKI_PREFIX}/summaries/source-{scope_slug}/docs/{doc_slug}/{vfolder}/L{level}/{filename}.md"
+            )
+        }
+        (SummaryTreeKind::Source, SummaryDiskLayout::Merge) => {
+            let filename = summary_filename(summary_id);
+            format!("{WIKI_PREFIX}/summaries/source-{scope_slug}/merge/L{level}/{filename}.md")
+        }
+        // Standard layout, or a non-Source tree kind — fall back to flat.
+        _ => summary_rel_path(tree_kind, scope_slug, level, summary_id),
+    }
+}
+
 /// Convert a summary id into the canonical on-disk basename stem (without
 /// `.md`).
 ///
@@ -338,6 +396,73 @@ fn truncate_at_char(s: &str, max_chars: usize) -> &str {
 mod tests {
     use super::*;
 
+    // ─── doc-aware layout tests ───────────────────────────────────────────────
+
+    #[test]
+    fn layout_doc_subtree_nests_under_docs_and_version() {
+        let p = summary_rel_path_with_layout(
+            SummaryTreeKind::Source,
+            "notion-conn1",
+            2,
+            "summary:1700000000000:L2-deadbeef",
+            SummaryDiskLayout::DocSubtree {
+                doc_slug: "notion-conn1-pageA",
+                version_ms: Some(1717500000000),
+            },
+        );
+        assert_eq!(
+            p,
+            "wiki/summaries/source-notion-conn1/docs/notion-conn1-pageA/v-1717500000000/L2/summary-1700000000000-L2-deadbeef.md"
+        );
+    }
+
+    #[test]
+    fn layout_doc_subtree_unversioned_folder() {
+        let p = summary_rel_path_with_layout(
+            SummaryTreeKind::Source,
+            "notion-conn1",
+            1,
+            "summary:1700000000000:L1-abcd0000",
+            SummaryDiskLayout::DocSubtree {
+                doc_slug: "notion-conn1-pageB",
+                version_ms: None,
+            },
+        );
+        assert!(
+            p.contains("/docs/notion-conn1-pageB/v-unversioned/L1/"),
+            "got {p}"
+        );
+    }
+
+    #[test]
+    fn layout_merge_tier_nests_under_merge() {
+        let p = summary_rel_path_with_layout(
+            SummaryTreeKind::Source,
+            "notion-conn1",
+            1000,
+            "summary:1700000000000:L1000-aaaa1111",
+            SummaryDiskLayout::Merge,
+        );
+        assert_eq!(
+            p,
+            "wiki/summaries/source-notion-conn1/merge/L1000/summary-1700000000000-L1000-aaaa1111.md"
+        );
+    }
+
+    #[test]
+    fn layout_standard_matches_flat_path() {
+        let id = "summary:1700000000000:L1-cccc2222";
+        let flat = summary_rel_path(SummaryTreeKind::Source, "slack-eng", 1, id);
+        let std_layout = summary_rel_path_with_layout(
+            SummaryTreeKind::Source,
+            "slack-eng",
+            1,
+            id,
+            SummaryDiskLayout::Standard,
+        );
+        assert_eq!(flat, std_layout);
+    }
+
     // ─── slugify tests ────────────────────────────────────────────────────────
 
     #[test]

src/openhuman/memory_store/content/read.rs

@@ -415,6 +415,8 @@ mod tests {
             sealed_at: ts,
             deleted: false,
             embedding: None,
+            doc_id: None,
+            version_ms: None,
         }
     }
 

src/openhuman/memory_store/traits.rs

@@ -232,6 +232,8 @@ mod tests {
             sealed_at: Utc::now(),
             deleted: false,
             embedding: None,
+            doc_id: None,
+            version_ms: None,
         };
         assert_eq!(node.memory_kind(), MemoryKind::Tree);
         assert_eq!(node.embeddable_text(), "summary body");

src/openhuman/memory_store/trees/store.rs

@@ -358,8 +358,9 @@ pub(crate) fn insert_summary_tx(
             entities_json, topics_json,
             time_range_start_ms, time_range_end_ms,
             score, sealed_at_ms, deleted, embedding,
-            content_path, content_sha256
-        ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16, ?17, ?18)",
+            content_path, content_sha256,
+            doc_id, version_ms
+        ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16, ?17, ?18, ?19, ?20)",
         params![
             node.id,
             node.tree_id,
@@ -379,6 +380,8 @@ pub(crate) fn insert_summary_tx(
             embedding_blob,
             content_path,
             content_sha256,
+            node.doc_id,
+            node.version_ms,
         ],
     )
     .with_context(|| format!("Failed to insert summary id={}", node.id))?;
@@ -694,7 +697,8 @@ pub fn get_summary(config: &Config, id: &str) -> Result<Option<SummaryNode>> {
                     child_ids_json, content, token_count,
                     entities_json, topics_json,
                     time_range_start_ms, time_range_end_ms,
-                    score, sealed_at_ms, deleted, embedding
+                    score, sealed_at_ms, deleted, embedding,
+                    doc_id, version_ms
                FROM mem_tree_summaries WHERE id = ?1",
         )?;
         let row = stmt
@@ -744,7 +748,8 @@ pub fn get_summaries_batch(
                         child_ids_json, content, token_count,
                         entities_json, topics_json,
                         time_range_start_ms, time_range_end_ms,
-                        score, sealed_at_ms, deleted, embedding
+                        score, sealed_at_ms, deleted, embedding,
+                        doc_id, version_ms
                    FROM mem_tree_summaries
                   WHERE id IN ({placeholders})"
             );
@@ -776,7 +781,8 @@ pub fn list_summaries_at_level(
                     child_ids_json, content, token_count,
                     entities_json, topics_json,
                     time_range_start_ms, time_range_end_ms,
-                    score, sealed_at_ms, deleted, embedding
+                    score, sealed_at_ms, deleted, embedding,
+                    doc_id, version_ms
                FROM mem_tree_summaries
               WHERE tree_id = ?1 AND level = ?2 AND deleted = 0
               ORDER BY sealed_at_ms ASC",
@@ -821,6 +827,8 @@ fn row_to_summary(row: &rusqlite::Row<'_>) -> rusqlite::Result<SummaryNode> {
     let sealed_ms: i64 = row.get(13)?;
     let deleted: i64 = row.get(14)?;
     let embedding_blob: Option<Vec<u8>> = row.get(15)?;
+    let doc_id: Option<String> = row.get(16)?;
+    let version_ms: Option<i64> = row.get(17)?;
 
     let tree_kind = TreeKind::parse(&tree_kind_s).map_err(|e| {
         rusqlite::Error::FromSqlConversionFailure(2, rusqlite::types::Type::Text, e.into())
@@ -863,6 +871,8 @@ fn row_to_summary(row: &rusqlite::Row<'_>) -> rusqlite::Result<SummaryNode> {
         sealed_at: ms_to_utc(sealed_ms)?,
         deleted: deleted != 0,
         embedding,
+        doc_id,
+        version_ms,
     })
 }
 

src/openhuman/memory_store/trees/store_tests.rs

@@ -43,6 +43,8 @@ fn sample_summary(id: &str, tree_id: &str, level: u32) -> SummaryNode {
         sealed_at: ts,
         deleted: false,
         embedding: None,
+        doc_id: None,
+        version_ms: None,
     }
 }
 

src/openhuman/memory_store/trees/types.rs

@@ -137,6 +137,25 @@ pub struct SummaryNode {
     /// dropping those rows to the bottom of semantic rerank results.
     #[serde(default)]
     pub embedding: Option<Vec<f32>>,
+    /// Document identity this node belongs to, for document source trees
+    /// (Notion etc.). `Some(source_id)` for nodes that live inside a single
+    /// document's per-doc subtree (its L1…doc-root chain); `None` for
+    /// merge-tier nodes (which summarise *across* documents) and for
+    /// chat/email source trees (which have no per-document structure).
+    ///
+    /// Together with [`Self::version_ms`] this lets retrieval resolve
+    /// "latest version per document" at read time: when a Notion page is
+    /// edited a new per-doc subtree is sealed with a higher `version_ms`,
+    /// and the older one is filtered out on traversal without ever being
+    /// rewritten or tombstoned.
+    #[serde(default)]
+    pub doc_id: Option<String>,
+    /// Document version this node was sealed for, as epoch-milliseconds
+    /// (Notion `last_edited_time`). `Some(_)` on per-doc subtree nodes,
+    /// `None` on merge-tier and non-document nodes. Read-time latest-wins
+    /// keeps `max(version_ms)` per [`Self::doc_id`].
+    #[serde(default)]
+    pub version_ms: Option<i64>,
 }
 
 /// Unsealed frontier at a given `(tree_id, level)`. One row per level per

src/openhuman/memory_sync/composio/providers/notion/ingest.rs

@@ -20,32 +20,45 @@
 //! accumulate under one source folder and one source tree instead of creating
 //! one graph source per page.
 //!
-//! ## Re-ingest of edited pages
+//! ## Page body content
 //!
-//! Notion pages mutate (the cursor advances by `last_edited_time`).
-//! Re-ingesting the same `(connection_id, page_id)` after the user edits
-//! the page would short-circuit on the pipeline's `already_ingested`
-//! gate — so the call site drops prior chunks for the same source_id
-//! via `delete_chunks_by_source` *before* re-ingest, mirroring the
-//! vault sync pattern in #2720. The provider's own
+//! `NOTION_FETCH_DATA` (the sync's listing call) returns page metadata +
+//! `properties` only — never the page body. The provider fetches each
+//! new/edited page's rendered body via `NOTION_GET_PAGE_MARKDOWN` and passes
+//! it here as `markdown_body`; [`render_page_body`] then emits the body as the
+//! primary content with the metadata JSON appended. Database/task rows have no
+//! body blocks (empty markdown) → metadata-only, which is correct since their
+//! data lives in `properties`.
+//!
+//! ## Re-ingest of edited pages (non-destructive, versioned)
+//!
+//! Notion pages mutate (`last_edited_time` advances). An edit is ingested
+//! **non-destructively**: `last_edited_time` becomes the document `version`,
+//! and the source-ingest gate is keyed by `{source_id}@{version}` (via
+//! [`ingest_pipeline::ingest_document_versioned`]) so a new revision is
+//! admitted *alongside* the prior one — the old chunks are NOT deleted
+//! (replacing the pre-versioning `delete_chunks_by_source` path). A
+//! [`JobKind::SealDocument`] job then builds this revision's per-document
+//! subtree (one versioned doc-root) and merges it into the connection tree;
+//! retrieval surfaces only the latest version per document. The provider's
 //! `SyncState::synced_ids` keyed by `{page_id}@{edited_time}` is the
-//! authoritative "have we seen this revision?" check; this module only
-//! runs when that says yes.
+//! authoritative "have we seen this revision?" check — unchanged pages are
+//! skipped before this module ever runs.
 //!
 //! ## Idempotency
 //!
-//! Chunk IDs are content-hashed inside the memory tree, so re-ingesting
-//! a previously-seen page is an UPSERT on the same chunk row — no
-//! duplicate chunks across syncs.
+//! Chunk IDs are content-hashed, so re-ingesting identical content is an
+//! UPSERT on the same chunk row — no duplicate chunks across syncs.
 
 use anyhow::Result;
 use chrono::{DateTime, Utc};
 use serde_json::Value;
 
 use crate::openhuman::config::Config;
-use crate::openhuman::memory::ingest_pipeline::{self, IngestResult};
-use crate::openhuman::memory_store::chunks::store::{delete_chunks_by_source, is_source_ingested};
-use crate::openhuman::memory_store::chunks::types::SourceKind;
+use crate::openhuman::memory::ingest_pipeline;
+use crate::openhuman::memory_queue::store::enqueue;
+use crate::openhuman::memory_queue::types::{NewJob, SealDocumentPayload};
+use crate::openhuman::memory_queue::wake_workers;
 use crate::openhuman::memory_sync::canonicalize::document::DocumentInput;
 
 /// Platform identifier embedded in the canonical document body header.
@@ -78,9 +91,25 @@ pub(crate) fn notion_source_scope(connection_id: &str) -> String {
 /// Composio response payload (not just the title) so the chunked content
 /// retains enough context for retrieval — Notion pages don't have a
 /// natural single-string canonical body the way Slack messages do.
-fn render_page_body(title: &str, page: &Value) -> String {
+/// Render the canonical document body for a Notion page.
+///
+/// When `markdown` is `Some` (the page's rendered body, fetched via
+/// `NOTION_GET_PAGE_MARKDOWN`), it is the primary content — the actual
+/// paragraphs, headings, lists, and *body tables* a free-form page contains.
+/// The `FETCH_DATA` metadata/properties JSON is appended after it so the
+/// page's structured fields (status, assignee, due date — the real data for
+/// database pages, which the markdown body does NOT include) stay searchable.
+///
+/// When `markdown` is `None` (no body fetched, or the page had none), we fall
+/// back to the metadata-only body — the prior behaviour.
+fn render_page_body(title: &str, page: &Value, markdown: Option<&str>) -> String {
     let pretty = serde_json::to_string_pretty(page).unwrap_or_else(|_| "{}".to_string());
-    format!("# {title}\n\n```json\n{pretty}\n```\n")
+    match markdown {
+        Some(md) if !md.trim().is_empty() => {
+            format!("# {title}\n\n{md}\n\n---\n\n```json\n{pretty}\n```\n")
+        }
+        _ => format!("# {title}\n\n```json\n{pretty}\n```\n"),
+    }
 }
 
 /// Parse a Notion `last_edited_time` (ISO 8601 / RFC 3339) into a
@@ -92,19 +121,24 @@ fn parse_edited_time(raw: Option<&str>) -> DateTime<Utc> {
         .unwrap_or_else(Utc::now)
 }
 
-/// Ingest one Notion page into the memory tree.
+/// Ingest one Notion page revision into the memory tree (non-destructive,
+/// versioned).
 ///
-/// Caller (the provider's `sync` loop) is responsible for the
-/// edit-detection / dedup state-machine (`SyncState::synced_ids` keyed
-/// by `{page_id}@{edited_time}`) — this function trusts that the call
-/// only happens for items the caller wants to admit.
+/// Caller (the provider's `sync` loop) owns the edit-detection / dedup
+/// state-machine (`SyncState::synced_ids` keyed by `{page_id}@{edited_time}`)
+/// — this function trusts it is only called for a revision the caller wants
+/// to admit.
 ///
-/// On content updates of an already-ingested source_id, drops the prior
-/// chunks first via `delete_chunks_by_source` so the pipeline's
-/// `already_ingested` gate doesn't short-circuit the new content. This
-/// mirrors the vault sync pattern in #2720.
+/// Unlike the pre-versioning behaviour (which deleted the prior chunks
+/// before re-ingesting), an edit is now **non-destructive**: the page's
+/// `last_edited_time` is used as the document version, the source-ingest gate
+/// is keyed by `{source_id}@{version}` (so a new revision is admitted
+/// alongside the old one), and a [`JobKind::SealDocument`] job is enqueued to
+/// build this revision's per-document subtree and merge its doc-root into the
+/// connection tree. Older revisions stay in place; the retrieval layer
+/// surfaces only the latest version per document.
 ///
-/// Returns the number of chunks the pipeline wrote.
+/// Returns the number of chunks the pipeline wrote for this revision.
 pub async fn ingest_page_into_memory_tree(
     config: &Config,
     connection_id: &str,
@@ -112,51 +146,17 @@ pub async fn ingest_page_into_memory_tree(
     title: &str,
     edited_time: Option<&str>,
     page: &Value,
+    markdown_body: Option<&str>,
 ) -> Result<usize> {
     let source_id = notion_source_id(connection_id, page_id);
-
-    // Re-sync of an edited page: drop prior chunks for the same source_id
-    // before re-ingest. Both calls are sync rusqlite I/O so they share one
-    // `spawn_blocking` hop.
-    //
-    // We gate `delete_chunks_by_source` behind `is_source_ingested` — the
-    // delete path uses a `source_kind = ?1` scan with Rust-side
-    // source-id filtering (see `store::delete_chunks_by_source_filter`),
-    // so on a first-time ingest of a never-seen page it would scan every
-    // Document-kind chunk just to find zero matches. `is_source_ingested`
-    // is an indexed PK lookup against `mem_tree_ingested_sources`, so it
-    // converts the common fresh-page case to one cheap `COUNT(*)` and only
-    // pays the scan cost on actual re-ingests of edited pages.
-    let cfg_for_blocking = config.clone();
-    let source_for_blocking = source_id.clone();
-    let removed = tokio::task::spawn_blocking(move || -> Result<usize> {
-        if is_source_ingested(
-            &cfg_for_blocking,
-            SourceKind::Document,
-            &source_for_blocking,
-        )? {
-            delete_chunks_by_source(
-                &cfg_for_blocking,
-                SourceKind::Document,
-                &source_for_blocking,
-            )
-        } else {
-            Ok(0)
-        }
-    })
-    .await
-    .map_err(|e| anyhow::anyhow!("delete-prior task join error: {e}"))??;
-    if removed > 0 {
-        tracing::debug!(
-            connection_id = %connection_id,
-            page_id = %page_id,
-            removed_chunks = removed,
-            "[composio:notion] ingest: re-ingest cleanup"
-        );
-    }
-
     let modified_at = parse_edited_time(edited_time);
-    let body = render_page_body(title, page);
+    // Document version = Notion `last_edited_time` in epoch-ms. Drives both
+    // the versioned ingest gate and the per-doc subtree's `version_ms` tag
+    // used for read-time latest-wins.
+    let version_ms = modified_at.timestamp_millis();
+    // Prefer the rendered page body (NOTION_GET_PAGE_MARKDOWN) when present;
+    // fall back to metadata-only when the caller didn't fetch it.
+    let body = render_page_body(title, page, markdown_body);
     let source_ref = Some(format!("notion://page/{page_id}"));
 
     let doc = DocumentInput {
@@ -170,37 +170,63 @@ pub async fn ingest_page_into_memory_tree(
     let owner = notion_source_scope(connection_id);
     let path_scope = Some(owner.clone());
 
-    match ingest_pipeline::ingest_document_with_scope(
-        config, &source_id, &owner, tags, doc, path_scope,
+    let result = ingest_pipeline::ingest_document_versioned(
+        config,
+        &source_id,
+        &owner,
+        tags,
+        doc,
+        path_scope,
+        Some(version_ms),
     )
     .await
-    {
-        Ok(IngestResult {
-            chunks_written,
-            already_ingested,
-            ..
-        }) => {
-            // The delete-first guard above prevents `already_ingested` on
-            // the normal update path. Seeing it here means the prior
-            // chunks were already absent (fresh ingest into a primed
-            // memory_tree) — fine, just log at debug.
-            tracing::debug!(
+    .map_err(|err| {
+        // `{err:#}` keeps the anyhow context chain so provider.rs's
+        // `tracing::warn!(error = %e)` shows the underlying cause.
+        anyhow::anyhow!("ingest_document failed for {source_id}: {err:#}")
+    })?;
+
+    if result.already_ingested {
+        // This exact revision was already ingested (provider dedup normally
+        // prevents reaching here). Nothing new to seal.
+        tracing::debug!(
+            connection_id = %connection_id,
+            page_id = %page_id,
+            version_ms,
+            "[composio:notion] ingest: revision already ingested — skipping seal"
+        );
+        return Ok(0);
+    }
+
+    // Enqueue the per-document seal: it rolls this revision's chunks up to a
+    // single doc-root (tagged with `version_ms`) and merges it into the
+    // connection tree. Forward-only — the prior revision's doc-root is left
+    // untouched; retrieval filters to the latest version per document.
+    if !result.chunk_ids.is_empty() {
+        let payload = SealDocumentPayload {
+            tree_scope: owner.clone(),
+            doc_id: source_id.clone(),
+            version_ms: Some(version_ms),
+            chunk_ids: result.chunk_ids.clone(),
+        };
+        match NewJob::seal_document(&payload).and_then(|job| enqueue(config, &job)) {
+            Ok(_) => wake_workers(),
+            Err(e) => tracing::warn!(
                 connection_id = %connection_id,
                 page_id = %page_id,
-                chunks_written,
-                already_ingested,
-                "[composio:notion] ingest: page persisted"
-            );
-            Ok(chunks_written)
+                "[composio:notion] ingest: failed to enqueue seal_document: {e:#}"
+            ),
         }
-        Err(err) => Err(anyhow::anyhow!(
-            // `{err:#}` (alternate formatter) bakes in the anyhow context
-            // chain so provider.rs's `tracing::warn!(error = %e)` doesn't
-            // strip the underlying cause (DB / embedding / persist failure)
-            // when it Displays the wrapped error.
-            "ingest_document failed for {source_id}: {err:#}"
-        )),
     }
+
+    tracing::debug!(
+        connection_id = %connection_id,
+        page_id = %page_id,
+        chunks_written = result.chunks_written,
+        version_ms,
+        "[composio:notion] ingest: page persisted (versioned)"
+    );
+    Ok(result.chunks_written)
 }
 
 #[cfg(test)]
@@ -295,13 +321,50 @@ mod tests {
     #[test]
     fn render_page_body_includes_title_header_and_pretty_json() {
         let page = json!({ "id": "p-1", "url": "https://notion.so/p1" });
-        let body = render_page_body("Phoenix plan", &page);
+        // Metadata-only (no markdown body): title header + pretty JSON.
+        let body = render_page_body("Phoenix plan", &page, None);
         assert!(body.starts_with("# Phoenix plan\n"));
         assert!(body.contains("```json\n"));
         assert!(body.contains("\"id\": \"p-1\""));
         assert!(body.contains("\"url\": \"https://notion.so/p1\""));
     }
 
+    /// With a markdown body present, render the body as the primary content
+    /// AND keep the metadata JSON appended after a `---` separator, so
+    /// free-form pages get real content while structured `properties` stay
+    /// searchable.
+    #[test]
+    fn render_page_body_merges_markdown_then_metadata() {
+        let page = json!({ "id": "p-1", "properties": { "Status": "Done" } });
+        let md = "## Heading\n\nReal body text with a list:\n- one\n- two";
+        let body = render_page_body("Phoenix plan", &page, Some(md));
+        assert!(body.starts_with("# Phoenix plan\n"));
+        assert!(body.contains("Real body text with a list"));
+        assert!(body.contains("\n---\n")); // separator before metadata
+        assert!(body.contains("```json")); // metadata still present
+        assert!(body.contains("\"Status\": \"Done\""));
+        // Body comes BEFORE the metadata block.
+        let md_pos = body.find("Real body text").unwrap();
+        let json_pos = body.find("```json").unwrap();
+        assert!(
+            md_pos < json_pos,
+            "markdown body must precede metadata json"
+        );
+    }
+
+    /// Empty/whitespace markdown falls back to metadata-only (a database row
+    /// with no body blocks returns empty markdown).
+    #[test]
+    fn render_page_body_empty_markdown_falls_back_to_metadata_only() {
+        let page = json!({ "id": "p-1" });
+        let body = render_page_body("Phoenix plan", &page, Some("   \n  "));
+        assert!(
+            !body.contains("\n---\n"),
+            "empty markdown → no body section"
+        );
+        assert!(body.contains("```json"));
+    }
+
     /// The #2885 regression test.
     ///
     /// Before this migration, Notion sync routed through
@@ -318,6 +381,7 @@ mod tests {
         use crate::openhuman::memory_store::chunks::store::{
             count_chunks, get_chunk_content_path, is_source_ingested, list_chunks, ListChunksQuery,
         };
+        use crate::openhuman::memory_store::chunks::types::SourceKind;
 
         let (_tmp, cfg) = test_config();
         let connection_id = "conn-test";
@@ -335,6 +399,7 @@ mod tests {
             "Phoenix migration plan",
             Some("2026-05-28T10:00:00.000Z"),
             &page,
+            None,
         )
         .await
         .expect("ingest_page_into_memory_tree");
@@ -382,29 +447,36 @@ mod tests {
             "source tag must not include page id: {body}"
         );
 
-        // Source registration.
+        // Source registration. Versioning keys the ingest gate by
+        // `{source_id}@{version_ms}` (version_ms = last_edited_time epoch-ms)
+        // so a later revision is admitted non-destructively — assert the
+        // versioned key is claimed, not the bare source_id.
+        let version_ms = parse_edited_time(Some("2026-05-28T10:00:00.000Z")).timestamp_millis();
+        let gate_key = format!("{expected}@{version_ms}");
         let cfg_for_blocking = cfg.clone();
-        let expected_for_task = expected.clone();
+        let gate_for_task = gate_key.clone();
         let registered = tokio::task::spawn_blocking(move || {
-            is_source_ingested(&cfg_for_blocking, SourceKind::Document, &expected_for_task)
+            is_source_ingested(&cfg_for_blocking, SourceKind::Document, &gate_for_task)
                 .unwrap_or(false)
         })
         .await
         .expect("source-check task join");
         assert!(
             registered,
-            "source_id {} must be registered in mem_tree_ingested_sources",
-            notion_source_id(connection_id, page_id)
+            "versioned gate key {gate_key} must be registered in mem_tree_ingested_sources"
         );
     }
 
     /// Re-ingesting an edited page (same `(connection_id, page_id)`,
-    /// different content) cleans up prior chunks and writes fresh ones —
-    /// the `delete_chunks_by_source` guard sidesteps the pipeline's
-    /// `already_ingested` short-circuit that would otherwise drop the
-    /// new revision.
+    /// different content + newer `last_edited_time`) is now
+    /// **non-destructive + versioned**: the new revision is admitted via the
+    /// `{source_id}@{version_ms}` gate key (not short-circuited), and the
+    /// prior revision's chunks are LEFT IN PLACE. Read-time latest-wins
+    /// surfaces only the newer version; the write path never deletes the old
+    /// one. (Replaces the pre-versioning destructive `delete_chunks_by_source`
+    /// behaviour.)
     #[tokio::test]
-    async fn re_ingesting_edited_page_replaces_prior_chunks() {
+    async fn re_ingesting_edited_page_keeps_both_versions() {
         use crate::openhuman::memory_store::chunks::store::count_chunks;
 
         let (_tmp, cfg) = test_config();
@@ -420,15 +492,16 @@ mod tests {
             "Phoenix plan v1",
             Some("2026-05-28T10:00:00.000Z"),
             &v1,
+            None,
         )
         .await
         .expect("first ingest");
         assert!(first > 0);
         let after_first = count_chunks(&cfg).expect("count after first");
 
-        // Re-ingest with different body — should NOT short-circuit, and
-        // chunk count should not double (prior chunks dropped, new ones
-        // written, net same per-page count for this body size).
+        // Re-ingest with different body + newer edit time — must NOT
+        // short-circuit (different version gate key) and must NOT delete the
+        // prior revision's chunks.
         let v2 = json!({
             "id": page_id,
             "object": "page",
@@ -443,6 +516,7 @@ mod tests {
             "Phoenix plan v2",
             Some("2026-05-29T10:00:00.000Z"),
             &v2,
+            None,
         )
         .await
         .expect("second ingest");
@@ -452,13 +526,11 @@ mod tests {
         );
         let after_second = count_chunks(&cfg).expect("count after second");
 
-        // The chunk count after the second ingest should equal the
-        // count after the first (replaced one revision with another),
-        // not double. Allow ±1 for any rounding in how the chunker
-        // splits subtly-different markdown.
+        // Non-destructive: the second revision's chunks are ADDED on top of
+        // the first revision's, so the total grows rather than staying flat.
         assert!(
-            after_second.abs_diff(after_first) <= 1,
-            "edited page must replace prior chunks, not append: \
+            after_second > after_first,
+            "edited page must keep both versions (non-destructive): \
              after_first={after_first} after_second={after_second}"
         );
     }

src/openhuman/memory_sync/composio/providers/notion/provider.rs

@@ -31,6 +31,11 @@ use futures::StreamExt;
 
 pub(crate) const ACTION_GET_ABOUT_ME: &str = "NOTION_GET_ABOUT_ME";
 pub(crate) const ACTION_FETCH_DATA: &str = "NOTION_FETCH_DATA";
+/// Per-page action that returns the page's rendered **body** as markdown
+/// (paragraphs, headings, lists, body tables). `FETCH_DATA` only returns
+/// metadata + properties; this fills in the actual document content for
+/// free-form pages. One request per page (budget-counted).
+pub(crate) const ACTION_GET_PAGE_MARKDOWN: &str = "NOTION_GET_PAGE_MARKDOWN";
 pub(crate) const ACTION_QUERY_DATABASE: &str = "NOTION_QUERY_DATABASE";
 pub(crate) const ACTION_SEARCH_NOTION_PAGE: &str = "NOTION_SEARCH_NOTION_PAGE";
 
@@ -298,6 +303,78 @@ impl ComposioProvider for NotionProvider {
             // ctx.max_items: clamp the dedup'd batch to the remaining budget before ingest.
             cap.clamp_batch(&mut pending);
 
+            // ── Step 4a.5: fetch each pending page's BODY markdown ──
+            // FETCH_DATA only returned metadata + properties. Pull the
+            // rendered page body (paragraphs, lists, body tables) per page so
+            // free-form documents ingest as real content (multi-chunk) rather
+            // than a single metadata chunk. One request per page — budget
+            // counted. Runs AFTER the depth-floor + max-items cap so we only
+            // fetch bodies for pages we'll actually ingest. On budget
+            // exhaustion or error we leave `markdown_body` None and ingest
+            // falls back to the metadata-only body.
+            for (idx, p) in pending.iter_mut().enumerate() {
+                if state.budget_exhausted() {
+                    tracing::info!(
+                        page = page_num,
+                        "[composio:notion] budget exhausted before body fetch — \
+                         remaining pages ingest metadata-only"
+                    );
+                    break;
+                }
+                match ctx
+                    .execute(
+                        ACTION_GET_PAGE_MARKDOWN,
+                        Some(json!({ "page_id": p.page_id })),
+                    )
+                    .await
+                {
+                    Ok(resp) => {
+                        state.record_requests(1);
+                        if resp.successful {
+                            p.markdown_body = sync::extract_page_markdown(&resp.data);
+                            // Empirical visibility (INFO so it shows at default
+                            // log level): on the first body fetch, log the
+                            // markdown length + the raw envelope keys so we can
+                            // confirm the response field / arg name on a live
+                            // run and refine if needed.
+                            if page_num == 0 && idx == 0 {
+                                tracing::info!(
+                                    page_id = %p.page_id,
+                                    markdown_chars =
+                                        p.markdown_body.as_ref().map(|s| s.len()).unwrap_or(0),
+                                    raw_keys = ?resp
+                                        .data
+                                        .as_object()
+                                        .map(|o| o.keys().cloned().collect::<Vec<_>>()),
+                                    "[composio:notion] GET_PAGE_MARKDOWN sample (empirical check)"
+                                );
+                            }
+                            if p.markdown_body.is_none() {
+                                tracing::warn!(
+                                    page_id = %p.page_id,
+                                    "[composio:notion] GET_PAGE_MARKDOWN returned no markdown \
+                                     field — metadata-only fallback"
+                                );
+                            }
+                        } else {
+                            tracing::warn!(
+                                page_id = %p.page_id,
+                                error = ?resp.error,
+                                "[composio:notion] GET_PAGE_MARKDOWN failed — metadata-only fallback"
+                            );
+                        }
+                    }
+                    Err(e) => {
+                        tracing::warn!(
+                            page_id = %p.page_id,
+                            error = %e,
+                            "[composio:notion] GET_PAGE_MARKDOWN execute error — \
+                             metadata-only fallback"
+                        );
+                    }
+                }
+            }
+
             // ── Step 4b: ingest queued pages (bounded concurrency) ──
             let ingestor = MemoryTreeIngestor {
                 config: ctx.config.as_ref(),
@@ -678,6 +755,11 @@ struct PendingIngest<'a> {
     title: String,
     edited_time: Option<String>,
     page: &'a Value,
+    /// Rendered page body (markdown) fetched per-page via
+    /// `NOTION_GET_PAGE_MARKDOWN` after dedupe, before ingest. `None` when the
+    /// body fetch was skipped (budget) or failed — ingest falls back to the
+    /// metadata-only body.
+    markdown_body: Option<String>,
 }
 
 /// Folded result of [`ingest_pending_buffered`]. Every field is
@@ -704,6 +786,7 @@ trait PageIngestor {
         title: &str,
         edited_time: Option<&str>,
         page: &Value,
+        markdown_body: Option<&str>,
     ) -> anyhow::Result<usize>;
 }
 
@@ -722,6 +805,7 @@ impl PageIngestor for MemoryTreeIngestor<'_> {
         title: &str,
         edited_time: Option<&str>,
         page: &Value,
+        markdown_body: Option<&str>,
     ) -> anyhow::Result<usize> {
         ingest_page_into_memory_tree(
             self.config,
@@ -730,6 +814,7 @@ impl PageIngestor for MemoryTreeIngestor<'_> {
             title,
             edited_time,
             page,
+            markdown_body,
         )
         .await
     }
@@ -793,6 +878,7 @@ fn select_pending<'a>(
             title,
             edited_time,
             page,
+            markdown_body: None,
         });
     }
     (pending, hit_cursor_boundary)
@@ -815,7 +901,13 @@ async fn ingest_pending_buffered<I: PageIngestor + Sync>(
         .into_iter()
         .map(|p| async move {
             let res = ingestor
-                .ingest(&p.page_id, &p.title, p.edited_time.as_deref(), p.page)
+                .ingest(
+                    &p.page_id,
+                    &p.title,
+                    p.edited_time.as_deref(),
+                    p.page,
+                    p.markdown_body.as_deref(),
+                )
                 .await;
             (p.sync_key, p.page_id, res)
         })
@@ -877,6 +969,7 @@ mod buffered_tests {
             _title: &str,
             _edited_time: Option<&str>,
             _page: &Value,
+            _markdown_body: Option<&str>,
         ) -> anyhow::Result<usize> {
             let now = self.in_flight.fetch_add(1, Ordering::SeqCst) + 1;
             self.peak.fetch_max(now, Ordering::SeqCst);
@@ -908,6 +1001,7 @@ mod buffered_tests {
                 title: format!("Notion: page {i}"),
                 edited_time: None,
                 page,
+                markdown_body: None,
             })
             .collect()
     }

src/openhuman/memory_sync/composio/providers/notion/sync.rs

@@ -22,6 +22,34 @@ pub(crate) fn extract_results(data: &Value) -> Vec<Value> {
     Vec::new()
 }
 
+/// Extract the rendered page body markdown from a `NOTION_GET_PAGE_MARKDOWN`
+/// response. Composio wraps action output in varying envelope shapes, so we
+/// try the common locations tolerantly and return the first non-empty string.
+/// Returns `None` if no markdown field is found (caller falls back to the
+/// metadata-only body and logs the raw shape for diagnosis).
+pub(crate) fn extract_page_markdown(data: &Value) -> Option<String> {
+    const PATHS: &[&str] = &[
+        "/markdown",
+        "/data/markdown",
+        "/data/response_data/markdown",
+        "/response_data/markdown",
+        "/data/content",
+        "/content",
+        "/data/markdown_content",
+        "/markdown_content",
+        "/text",
+        "/data/text",
+    ];
+    for p in PATHS {
+        if let Some(s) = data.pointer(p).and_then(Value::as_str) {
+            if !s.trim().is_empty() {
+                return Some(s.to_string());
+            }
+        }
+    }
+    None
+}
+
 /// Extract the Notion pagination cursor (for `start_cursor` on the
 /// next request).
 pub(crate) fn extract_notion_cursor(data: &Value) -> Option<String> {
@@ -94,6 +122,37 @@ mod tests {
         assert_eq!(results.len(), 1);
     }
 
+    #[test]
+    fn extract_page_markdown_reads_top_level_field() {
+        // Matches the live GET_PAGE_MARKDOWN envelope observed empirically:
+        // {id, markdown, object, request_id, truncated, unknown_block_ids}.
+        let data = json!({
+            "id": "p1",
+            "markdown": "# Heading\n\nbody text",
+            "object": "page",
+            "truncated": false,
+        });
+        assert_eq!(
+            extract_page_markdown(&data).as_deref(),
+            Some("# Heading\n\nbody text")
+        );
+    }
+
+    #[test]
+    fn extract_page_markdown_reads_nested_envelope() {
+        let data = json!({ "data": { "markdown": "nested body" } });
+        assert_eq!(extract_page_markdown(&data).as_deref(), Some("nested body"));
+    }
+
+    #[test]
+    fn extract_page_markdown_none_for_empty_or_missing() {
+        // Empty markdown (a DB row with no body blocks) → None → metadata-only.
+        assert_eq!(extract_page_markdown(&json!({ "markdown": "" })), None);
+        assert_eq!(extract_page_markdown(&json!({ "markdown": "   " })), None);
+        // No markdown field at all → None.
+        assert_eq!(extract_page_markdown(&json!({ "id": "p1" })), None);
+    }
+
     #[test]
     fn extract_results_from_top_level() {
         let data = json!({"results": [{"id": "a"}, {"id": "b"}]});

src/openhuman/memory_tree/ingest.rs

@@ -133,6 +133,8 @@ pub async fn ingest_summary(
         sealed_at: now,
         deleted: false,
         embedding: None,
+        doc_id: None,
+        version_ms: None,
     };
 
     // Persist summary + update buffer in one transaction.

src/openhuman/memory_tree/retrieval/drill_down.rs

@@ -18,7 +18,7 @@
 //! - Leaves have no children; drilling into a leaf id returns empty.
 //! - `limit` is optional; when set, it truncates the final (reranked) output.
 
-use std::collections::HashSet;
+use std::collections::{HashMap, HashSet};
 
 use anyhow::Result;
 
@@ -217,6 +217,22 @@ fn walk_with_embeddings(
     let mut current_level: Vec<String> = start_children;
     let mut depth: u32 = 1;
 
+    // Latest-version-per-document filter (document source trees, e.g. Notion).
+    // A document's chunks roll up to a per-doc subtree whose root carries
+    // `(doc_id, version_ms)`; editing a page seals a NEW doc-root (higher
+    // `version_ms`) alongside the old one, so the merge tier reaches both. We
+    // surface only the newest revision: as the walk encounters doc-roots we
+    // track `max(version_ms)` per `doc_id` and skip any doc-root below that
+    // max (and therefore its whole stale subtree). Nothing is mutated on disk
+    // — superseded revisions simply never appear in results. Non-document
+    // nodes (doc_id == None) are unaffected.
+    let mut max_version_by_doc: HashMap<String, i64> = HashMap::new();
+    // Doc-roots already surfaced, to dedup at the winning version: if a
+    // `SealDocument` job partially committed then retried, it can mint a second
+    // doc-root for the SAME `(doc_id, version_ms)`. Emit only the first one per
+    // doc_id so a duplicate revision never double-surfaces.
+    let mut emitted_docs: HashSet<String> = HashSet::new();
+
     while !current_level.is_empty() && depth <= max_depth {
         log::trace!(
             "[retrieval::drill_down] level depth={} ids={}",
@@ -230,6 +246,29 @@ fn walk_with_embeddings(
         //    tried as chunks below.
         let mut summary_by_id = get_summaries_batch(config, &current_level)?;
 
+        // Update the per-document latest-version map with any doc-roots on
+        // THIS level before walking it, so two revisions of the same document
+        // sitting side-by-side (the common case — both are merge-tier leaves
+        // at the same depth) resolve to the newer one regardless of walk
+        // order. A doc-root is a summary with `doc_id` set; `version_ms`
+        // defaults to i64::MIN so a (legacy) untagged doc-root never wins over
+        // a tagged one.
+        for id in &current_level {
+            if let Some(s) = summary_by_id.get(id) {
+                if let Some(doc_id) = s.doc_id.as_deref() {
+                    let v = s.version_ms.unwrap_or(i64::MIN);
+                    max_version_by_doc
+                        .entry(doc_id.to_string())
+                        .and_modify(|cur| {
+                            if v > *cur {
+                                *cur = v;
+                            }
+                        })
+                        .or_insert(v);
+                }
+            }
+        }
+
         // 2) Distinct tree_ids referenced by this level's summaries —
         //    dedup is purely to avoid redundant DB params (the per-id
         //    walk below routes each summary to its own scope via the
@@ -284,6 +323,29 @@ fn walk_with_embeddings(
         };
         for id in &current_level {
             if let Some(mut summary) = summary_by_id.remove(id) {
+                // Latest-wins: skip a doc-root that a newer revision of the
+                // same document supersedes. Its subtree is not expanded, so
+                // the stale revision's chunks never surface.
+                if let Some(doc_id) = summary.doc_id.as_deref() {
+                    let v = summary.version_ms.unwrap_or(i64::MIN);
+                    if max_version_by_doc.get(doc_id).is_some_and(|&max| v < max) {
+                        log::debug!(
+                            "[retrieval::drill_down] skipping superseded doc-root \
+                             doc_id={doc_id} version_ms={v} (latest is newer)"
+                        );
+                        continue;
+                    }
+                    // Dedup duplicates at the winning version (e.g. a retried
+                    // SealDocument that minted a second doc-root for the same
+                    // (doc_id, version_ms)) — surface only the first.
+                    if !emitted_docs.insert(doc_id.to_string()) {
+                        log::debug!(
+                            "[retrieval::drill_down] skipping duplicate doc-root \
+                             doc_id={doc_id} version_ms={v} (already surfaced)"
+                        );
+                        continue;
+                    }
+                }
                 let scope = tree_by_id
                     .get(&summary.tree_id)
                     .map(|t| t.scope.clone())
@@ -453,6 +515,131 @@ mod tests {
         assert!(out.is_empty());
     }
 
+    /// Read-time latest-wins: a merge root referencing two per-doc roots of
+    /// the SAME document (v1 < v2) must surface only the newer revision's
+    /// subtree; the superseded doc-root and its chunk are filtered out and
+    /// never traversed — without anything being deleted on disk.
+    #[tokio::test]
+    async fn drill_down_surfaces_only_latest_doc_version() {
+        use crate::openhuman::memory_store::chunks::store::{upsert_chunks, with_connection};
+        use crate::openhuman::memory_store::trees::types::{SummaryNode, Tree, TreeStatus};
+        use crate::openhuman::memory_tree::tree::store as tree_store;
+
+        let (_tmp, cfg) = test_config();
+        let ts = Utc::now();
+
+        let tree = Tree {
+            id: "test:notion-tree".into(),
+            kind: TreeKind::Source,
+            scope: "notion:conn1".into(),
+            root_id: Some("s:merge:root".into()),
+            max_level: 1000,
+            status: TreeStatus::Active,
+            created_at: ts,
+            last_sealed_at: Some(ts),
+        };
+
+        let mk_chunk = |content: &str| Chunk {
+            id: chunk_id(SourceKind::Document, "notion:conn1:pageA", 0, content),
+            content: content.to_string(),
+            metadata: Metadata {
+                source_kind: SourceKind::Document,
+                source_id: "notion:conn1:pageA".into(),
+                owner: "notion:conn1".into(),
+                timestamp: ts,
+                time_range: (ts, ts),
+                tags: vec!["notion".into()],
+                source_ref: Some(SourceRef::new("notion://page/pageA")),
+                path_scope: Some("notion:conn1".into()),
+            },
+            token_count: 10,
+            seq_in_source: 0,
+            created_at: ts,
+            partial_message: false,
+        };
+        // Distinct content → distinct chunk ids (content is hashed in).
+        let chunk_v1 = mk_chunk("old version body");
+        let chunk_v2 = mk_chunk("new version body");
+        upsert_chunks(&cfg, &[chunk_v1.clone(), chunk_v2.clone()]).unwrap();
+
+        let mk_root = |id: &str, version: i64, child: &str| SummaryNode {
+            id: id.into(),
+            tree_id: tree.id.clone(),
+            tree_kind: TreeKind::Source,
+            level: 1,
+            parent_id: Some("s:merge:root".into()),
+            child_ids: vec![child.to_string()],
+            content: format!("doc-root v{version}"),
+            token_count: 5,
+            entities: vec![],
+            topics: vec![],
+            time_range_start: ts,
+            time_range_end: ts,
+            score: 0.5,
+            sealed_at: ts,
+            deleted: false,
+            embedding: None,
+            doc_id: Some("notion:conn1:pageA".into()),
+            version_ms: Some(version),
+        };
+        let v1_root = mk_root("s:docA:v1", 100, &chunk_v1.id);
+        let v2_root = mk_root("s:docA:v2", 200, &chunk_v2.id);
+
+        let merge_root = SummaryNode {
+            id: "s:merge:root".into(),
+            tree_id: tree.id.clone(),
+            tree_kind: TreeKind::Source,
+            level: 1000,
+            parent_id: None,
+            child_ids: vec![v1_root.id.clone(), v2_root.id.clone()],
+            content: "merge root".into(),
+            token_count: 5,
+            entities: vec![],
+            topics: vec![],
+            time_range_start: ts,
+            time_range_end: ts,
+            score: 0.5,
+            sealed_at: ts,
+            deleted: false,
+            embedding: None,
+            doc_id: None,
+            version_ms: None,
+        };
+
+        with_connection(&cfg, |conn| {
+            let tx = conn.unchecked_transaction()?;
+            tree_store::insert_tree_conn(&tx, &tree)?;
+            tree_store::insert_summary_tx(&tx, &v1_root, None, "test")?;
+            tree_store::insert_summary_tx(&tx, &v2_root, None, "test")?;
+            tree_store::insert_summary_tx(&tx, &merge_root, None, "test")?;
+            tx.commit()?;
+            Ok(())
+        })
+        .unwrap();
+
+        let out = drill_down(&cfg, "s:merge:root", 3, None, None)
+            .await
+            .unwrap();
+        let ids: Vec<&str> = out.iter().map(|h| h.node_id.as_str()).collect();
+
+        assert!(
+            ids.contains(&"s:docA:v2"),
+            "latest doc-root must surface; got {ids:?}"
+        );
+        assert!(
+            ids.contains(&chunk_v2.id.as_str()),
+            "latest version's chunk must surface; got {ids:?}"
+        );
+        assert!(
+            !ids.contains(&"s:docA:v1"),
+            "superseded doc-root must be filtered; got {ids:?}"
+        );
+        assert!(
+            !ids.contains(&chunk_v1.id.as_str()),
+            "superseded version's chunk must not surface; got {ids:?}"
+        );
+    }
+
     #[tokio::test]
     async fn summary_drills_to_leaves_at_depth_one() {
         let (_tmp, cfg) = test_config();
@@ -614,6 +801,8 @@ mod tests {
             sealed_at: ts,
             deleted: false,
             embedding: None,
+            doc_id: None,
+            version_ms: None,
         };
         let l1_b = SummaryNode {
             id: "s:L1:b".into(),
@@ -730,6 +919,8 @@ mod tests {
             sealed_at: ts,
             deleted: false,
             embedding: None,
+            doc_id: None,
+            version_ms: None,
         };
         let l1_b = SummaryNode {
             id: "s:L1:b".into(),

src/openhuman/memory_tree/tree/bucket_seal.rs

@@ -42,7 +42,11 @@ use rusqlite::Transaction;
 
 use crate::openhuman::config::Config;
 use crate::openhuman::memory_store::chunks::store::with_connection;
-use crate::openhuman::memory_store::content::{atomic::stage_summary, SummaryComposeInput};
+use crate::openhuman::memory_store::content::{
+    atomic::stage_summary_with_layout,
+    paths::{slugify_source_id, SummaryDiskLayout},
+    SummaryComposeInput,
+};
 use crate::openhuman::memory_store::trees::types::{
     Buffer, SummaryNode, Tree, TreeKind, INPUT_TOKEN_BUDGET, OUTPUT_TOKEN_BUDGET, SUMMARY_FANOUT,
 };
@@ -50,7 +54,7 @@ use crate::openhuman::memory_tree::score::embed::build_write_embedder;
 use crate::openhuman::memory_tree::score::extract::EntityExtractor;
 use crate::openhuman::memory_tree::score::resolver::canonicalise;
 use crate::openhuman::memory_tree::summarise::{
-    fallback_summary, summarise, SummaryContext, SummaryInput,
+    fallback_summary, summarise, SummaryContext, SummaryInput, SummaryOutput,
 };
 use crate::openhuman::memory_tree::tree::factory::TreeFactory;
 use crate::openhuman::memory_tree::tree::registry::new_summary_id;
@@ -533,6 +537,11 @@ pub(crate) async fn seal_one_level(
         sealed_at: now,
         deleted: false,
         embedding,
+        // Generic seal path (chat/email source trees + the cross-document
+        // merge tier) is document-agnostic. The per-document subtree seal
+        // (Notion) sets these via its own seal entrypoint in Task #2.
+        doc_id: None,
+        version_ms: None,
     };
 
     crate::core::event_bus::publish_global(
@@ -651,12 +660,22 @@ pub(crate) async fn seal_one_level(
              continuing seal without vault defaults"
         );
     }
-    let staged = stage_summary(&content_root, &compose_input, &scope_slug).with_context(|| {
-        format!(
-            "stage_summary failed for {}; seal aborted, buffer stays unsealed for retry",
-            node.id
-        )
-    })?;
+    // Merge-tier nodes (document source trees, level ≥ MERGE_LEVEL_BASE) land
+    // under `source-<scope>/merge/`; everything else (chat/email + the
+    // per-doc subtree is sealed via seal_explicit_children, not here) uses the
+    // flat layout.
+    let layout = if node.level >= MERGE_LEVEL_BASE {
+        SummaryDiskLayout::Merge
+    } else {
+        SummaryDiskLayout::Standard
+    };
+    let staged = stage_summary_with_layout(&content_root, &compose_input, &scope_slug, layout)
+        .with_context(|| {
+            format!(
+                "stage_summary failed for {}; seal aborted, buffer stays unsealed for retry",
+                node.id
+            )
+        })?;
     log::debug!(
         "[tree::bucket_seal] staged summary {} → {}",
         node.id,
@@ -925,6 +944,431 @@ fn hydrate_summary_inputs(config: &Config, summary_ids: &[String]) -> Result<Vec
     Ok(out)
 }
 
+// ── Document-aware sealing (Notion etc.) ────────────────────────────────
+//
+// Document source trees keep ONE physical `mem_tree_trees` row per
+// connection (e.g. `notion:{connection_id}`), but inside it each document
+// rolls up to its own **doc-root** summary, and those doc-roots merge into
+// the connection root. To get that shape without re-keying the shared
+// `(tree, level)` buffer — the exact path chat/email seal through — the
+// per-document subtree is built as an **isolated side-cascade**
+// ([`seal_document_subtree`]) that never touches a shared buffer or the
+// tree root. Only the cross-document **merge tier** uses the shared buffer
+// + the existing [`cascade_all_from`] engine, starting at
+// [`MERGE_LEVEL_BASE`].
+//
+// Versioning is forward-only: editing a Notion page calls
+// [`seal_document_subtree`] again with a higher `version_ms`, producing a
+// *new* doc-root that is appended to the merge buffer alongside the old
+// one. Nothing is rewritten or tombstoned; retrieval keeps `max(version_ms)`
+// per `doc_id` at read time (see the retrieval layer).
+
+/// Level offset where the cross-document merge tier lives inside a document
+/// source tree.
+///
+/// Per-document subtrees occupy small levels (1, 2, …) and are built by
+/// [`seal_document_subtree`] as a side-cascade that never enters a shared
+/// `(tree, level)` buffer. The merge tier — which summarises *across*
+/// documents — uses the shared buffer and the existing cascade engine,
+/// starting here. The wide gap guarantees per-doc nodes (small levels) and
+/// merge nodes (≥ `MERGE_LEVEL_BASE`) can never collide on `(tree, level)`,
+/// and keeps `Tree.root_id` / `max_level` pointing at a merge node.
+pub const MERGE_LEVEL_BASE: u32 = 1_000;
+
+/// Hard cap on how many children one per-document summary fans in, so a
+/// single huge document can't produce a doc-root with thousands of direct
+/// children. Independent of [`SUMMARY_FANOUT`] (which gates the merge tier).
+const DOC_SUBTREE_MAX_FANIN: usize = 32;
+
+/// Build (or re-build, for a new version) one document's subtree and merge
+/// its doc-root into the connection tree.
+///
+/// `doc_id` is the document identity (the chunk `source_id`, e.g.
+/// `notion:{conn}:{page_id}`); `version_ms` is the document version
+/// (`last_edited_time` epoch-ms). `chunk_ids` are this version's leaf chunk
+/// ids, already persisted in `mem_tree_chunks`.
+///
+/// Steps:
+/// 1. Cascade `chunk_ids` upward (token-budget batches at L0, count batches
+///    above) until a **single doc-root** remains — force-sealed even for a
+///    one-chunk document so it always surfaces as a doc-root, never loose
+///    leaves. Every node is tagged `(doc_id, version_ms)`.
+/// 2. Append the doc-root to the connection tree's merge buffer at
+///    [`MERGE_LEVEL_BASE`] and run the existing cascade so it folds into the
+///    connection root once `SUMMARY_FANOUT` doc-roots accumulate.
+///
+/// Returns the doc-root summary id. Idempotent re-runs for the *same*
+/// `(doc_id, version_ms, chunk_ids)` produce a new doc-root (new ids); the
+/// caller (Notion sync) only invokes this when a new revision is admitted.
+pub async fn seal_document_subtree(
+    config: &Config,
+    tree: &Tree,
+    doc_id: &str,
+    version_ms: Option<i64>,
+    chunk_ids: &[String],
+    strategy: &LabelStrategy,
+) -> Result<String> {
+    if chunk_ids.is_empty() {
+        anyhow::bail!(
+            "[tree::bucket_seal] seal_document_subtree: empty chunk set tree_id={} doc_id={}",
+            tree.id,
+            doc_id
+        );
+    }
+    log::debug!(
+        "[tree::bucket_seal] seal_document_subtree tree_id={} doc_id={} version_ms={:?} chunks={}",
+        tree.id,
+        doc_id,
+        version_ms,
+        chunk_ids.len()
+    );
+
+    // 1. Per-document side-cascade to a single doc-root.
+    let mut current_level: u32 = 0;
+    let mut current_ids: Vec<String> = chunk_ids.to_vec();
+    let mut doc_root: Option<SummaryNode> = None;
+
+    loop {
+        let batches = if current_level == 0 {
+            batch_leaves_by_token_budget(config, &current_ids)?
+        } else {
+            batch_by_count(&current_ids, DOC_SUBTREE_MAX_FANIN)
+        };
+
+        let mut next_ids: Vec<String> = Vec::with_capacity(batches.len());
+        for batch in &batches {
+            let node = seal_explicit_children(
+                config,
+                tree,
+                current_level,
+                batch,
+                Some(doc_id),
+                version_ms,
+                strategy,
+            )
+            .await?;
+            next_ids.push(node.id.clone());
+            doc_root = Some(node);
+        }
+
+        current_level += 1;
+        current_ids = next_ids;
+        if current_ids.len() <= 1 {
+            break;
+        }
+    }
+
+    let doc_root = doc_root.ok_or_else(|| {
+        anyhow::anyhow!(
+            "[tree::bucket_seal] seal_document_subtree produced no doc-root tree_id={} doc_id={}",
+            tree.id,
+            doc_id
+        )
+    })?;
+    log::debug!(
+        "[tree::bucket_seal] doc-root sealed tree_id={} doc_id={} root_id={} level={}",
+        tree.id,
+        doc_id,
+        doc_root.id,
+        doc_root.level
+    );
+
+    // 2. Feed the doc-root into the cross-document merge tier and cascade
+    //    using the untouched shared engine.
+    append_to_buffer(
+        config,
+        &tree.id,
+        MERGE_LEVEL_BASE,
+        &doc_root.id,
+        doc_root.token_count as i64,
+        doc_root.time_range_start,
+    )?;
+    let merge_sealed = cascade_all_from(config, tree, MERGE_LEVEL_BASE, None, strategy).await?;
+    log::debug!(
+        "[tree::bucket_seal] merge cascade tree_id={} doc_id={} merge_sealed={}",
+        tree.id,
+        doc_id,
+        merge_sealed.len()
+    );
+
+    Ok(doc_root.id)
+}
+
+/// Greedily batch leaf chunk ids so each batch stays under
+/// [`INPUT_TOKEN_BUDGET`] (and at most [`DOC_SUBTREE_MAX_FANIN`] children).
+/// A single oversized chunk forms its own batch.
+fn batch_leaves_by_token_budget(config: &Config, chunk_ids: &[String]) -> Result<Vec<Vec<String>>> {
+    use crate::openhuman::memory_store::chunks::store::get_chunk;
+
+    let mut batches: Vec<Vec<String>> = Vec::new();
+    let mut current: Vec<String> = Vec::new();
+    let mut token_sum: i64 = 0;
+
+    for id in chunk_ids {
+        let tokens = match get_chunk(config, id)? {
+            Some(c) => c.token_count as i64,
+            None => {
+                log::warn!(
+                    "[tree::bucket_seal] batch_leaves_by_token_budget: missing chunk {id} — skipping"
+                );
+                continue;
+            }
+        };
+        let would_exceed = token_sum + tokens > INPUT_TOKEN_BUDGET as i64
+            || current.len() >= DOC_SUBTREE_MAX_FANIN;
+        if would_exceed && !current.is_empty() {
+            batches.push(std::mem::take(&mut current));
+            token_sum = 0;
+        }
+        current.push(id.clone());
+        token_sum += tokens;
+    }
+    if !current.is_empty() {
+        batches.push(current);
+    }
+    if batches.is_empty() {
+        // All chunks were missing — surface as one empty batch caller rejects.
+        anyhow::bail!("[tree::bucket_seal] batch_leaves_by_token_budget: no resolvable chunks");
+    }
+    Ok(batches)
+}
+
+/// Split ids into fixed-size batches of at most `max` (used above L0 in the
+/// per-document cascade).
+fn batch_by_count(ids: &[String], max: usize) -> Vec<Vec<String>> {
+    ids.chunks(max.max(1)).map(|c| c.to_vec()).collect()
+}
+
+/// Seal an **explicit** set of child ids into one summary at `level + 1`,
+/// tagging it with `doc_id` / `version_ms`. Unlike [`seal_one_level`] this
+/// does NOT touch any shared `(tree, level)` buffer and does NOT advance the
+/// tree root/max_level — it is the per-document side-cascade primitive. It
+/// reuses the same hydrate → summarise → label → embed → stage → persist
+/// pipeline so doc-subtree summaries are indistinguishable from regular
+/// summaries except for their `doc_id` / `version_ms` tags.
+async fn seal_explicit_children(
+    config: &Config,
+    tree: &Tree,
+    level: u32,
+    child_ids: &[String],
+    doc_id: Option<&str>,
+    version_ms: Option<i64>,
+    strategy: &LabelStrategy,
+) -> Result<SummaryNode> {
+    let target_level = level + 1;
+    let inputs = hydrate_inputs(config, level, child_ids)?;
+    if inputs.is_empty() {
+        anyhow::bail!(
+            "[tree::bucket_seal] seal_explicit_children: empty inputs tree_id={} level={}",
+            tree.id,
+            level
+        );
+    }
+
+    let time_range_start = inputs
+        .iter()
+        .map(|i| i.time_range_start)
+        .min()
+        .unwrap_or_else(Utc::now);
+    let time_range_end = inputs
+        .iter()
+        .map(|i| i.time_range_end)
+        .max()
+        .unwrap_or_else(Utc::now);
+    let score = inputs
+        .iter()
+        .map(|i| i.score)
+        .fold(f32::NEG_INFINITY, f32::max)
+        .max(0.0);
+
+    let ctx = SummaryContext {
+        tree_id: &tree.id,
+        tree_kind: tree.kind,
+        target_level,
+        token_budget: OUTPUT_TOKEN_BUDGET,
+    };
+    // Single-input passthrough: if a doc rolls up from exactly one node that
+    // already fits the summary budget (the common case — a Notion page that is
+    // a single chunk), there is nothing to summarise. Emit the input verbatim
+    // as the doc-root content and SKIP the LLM call entirely. The doc-root is
+    // still a real summary node (so versioning, the merge tier, and the
+    // read-time latest-version filter all keep working uniformly) — it just
+    // isn't a redundant paraphrase of one chunk, and costs no inference.
+    // A single oversized input still goes through the summariser (it genuinely
+    // needs compression).
+    let output = if inputs.len() == 1 && inputs[0].token_count <= OUTPUT_TOKEN_BUDGET {
+        log::debug!(
+            "[tree::bucket_seal] doc-subtree passthrough (1 input, no LLM) tree_id={} doc_id={:?} level={}",
+            tree.id,
+            doc_id,
+            level
+        );
+        let only = &inputs[0];
+        SummaryOutput {
+            content: only.content.clone(),
+            token_count: only.token_count,
+            entities: Vec::new(),
+            topics: Vec::new(),
+            input_tokens: 0,
+            output_tokens: 0,
+            charged_amount_usd: None,
+        }
+    } else {
+        match summarise(config, &inputs, &ctx).await {
+            Ok(o) => o,
+            Err(e) => {
+                log::warn!(
+                    "[tree::bucket_seal] doc-subtree summarise failed tree_id={} doc_id={:?} level={}: {e:#} — fallback",
+                    tree.id, doc_id, level,
+                );
+                fallback_summary(&inputs, ctx.token_budget)
+            }
+        }
+    };
+
+    let (node_entities, node_topics) = resolve_labels(strategy, &inputs, &output.content).await?;
+
+    // Embed before any write so a failure aborts cleanly — same contract as
+    // seal_one_level. No-provider configs seal embedding-less.
+    let embed_input = truncate_for_embed(&output.content, 1_000);
+    let embedding: Option<Vec<f32>> =
+        match build_write_embedder(config).context("build embedder during doc-subtree seal")? {
+            None => None,
+            Some(embedder) => {
+                let v = embedder.embed(&embed_input).await.map_err(|e| {
+                    let failure = crate::openhuman::memory_tree::health::classify_embed_error(&e);
+                    anyhow::Error::new(failure).context(format!(
+                        "embed doc-subtree summary tree_id={} level={}: {e:#}",
+                        tree.id, level
+                    ))
+                })?;
+                crate::openhuman::memory_tree::score::embed::pack_checked(&v).context(format!(
+                    "doc-subtree embed dim check tree_id={} level={}",
+                    tree.id, level
+                ))?;
+                Some(v)
+            }
+        };
+
+    let now = Utc::now();
+    let summary_id = new_summary_id(target_level);
+    let node = SummaryNode {
+        id: summary_id.clone(),
+        tree_id: tree.id.clone(),
+        tree_kind: tree.kind,
+        level: target_level,
+        parent_id: None,
+        child_ids: child_ids.to_vec(),
+        content: output.content,
+        token_count: output.token_count,
+        entities: node_entities,
+        topics: node_topics,
+        time_range_start,
+        time_range_end,
+        score,
+        sealed_at: now,
+        deleted: false,
+        embedding,
+        doc_id: doc_id.map(|s| s.to_string()),
+        version_ms,
+    };
+
+    // Stage the .md file before opening the write tx (same fail-fast as
+    // seal_one_level). Doc-subtree nodes land under
+    // `source-<scope>/docs/<doc_slug>/v-<version_ms>/…` so the vault mirrors
+    // the logical shape. Wikilink overrides are left unset.
+    let tree_factory = TreeFactory::from_tree(tree);
+    let summary_tree_kind = tree_factory.summary_tree_kind();
+    let scope_slug = tree_factory.scope_slug();
+    let compose_input = SummaryComposeInput {
+        summary_id: &node.id,
+        tree_kind: summary_tree_kind,
+        tree_id: &node.tree_id,
+        tree_scope: &tree.scope,
+        level: node.level,
+        child_ids: &node.child_ids,
+        child_basenames: None,
+        child_count: node.child_ids.len(),
+        time_range_start: node.time_range_start,
+        time_range_end: node.time_range_end,
+        sealed_at: node.sealed_at,
+        body: &node.content,
+    };
+    let content_root = config.memory_tree_content_root();
+    let doc_slug = doc_id.map(slugify_source_id);
+    let layout = match doc_slug.as_deref() {
+        Some(slug) => SummaryDiskLayout::DocSubtree {
+            doc_slug: slug,
+            version_ms,
+        },
+        None => SummaryDiskLayout::Standard,
+    };
+    let staged = stage_summary_with_layout(&content_root, &compose_input, &scope_slug, layout)
+        .with_context(|| {
+            format!(
+                "stage_summary failed for doc-subtree node {}; seal aborted",
+                node.id
+            )
+        })?;
+
+    // Persist the summary row + backlink children — NO buffer / tree-root
+    // mutation (those belong to the merge tier).
+    let node_for_tx = node.clone();
+    let level_for_tx = level;
+    let summary_id_for_tx = summary_id.clone();
+    let signature = crate::openhuman::memory_store::chunks::store::tree_active_signature(config);
+    with_connection(config, move |conn| {
+        let tx = conn.unchecked_transaction()?;
+        store::insert_summary_tx(&tx, &node_for_tx, Some(&staged), &signature)?;
+        crate::openhuman::memory_tree::score::store::index_summary_entity_ids_tx(
+            &tx,
+            &node_for_tx.entities,
+            &node_for_tx.id,
+            node_for_tx.score,
+            now.timestamp_millis(),
+            Some(&node_for_tx.tree_id),
+        )?;
+        for child_id in &node_for_tx.child_ids {
+            if level_for_tx == 0 {
+                // Unconditional re-point (no `IS NULL` guard): a byte-identical
+                // body chunk reused across doc versions upserts to the SAME row
+                // (content-addressed id), so its single `parent_summary_id` must
+                // follow the newest version. Doc subtrees seal newest-last, so
+                // last-write-wins leaves the backlink on the latest doc-root —
+                // the version retrieval surfaces — instead of stranding it on
+                // the first (now-superseded) version's summary.
+                tx.execute(
+                    "UPDATE mem_tree_chunks SET parent_summary_id = ?1 \
+                       WHERE id = ?2",
+                    rusqlite::params![&summary_id_for_tx, child_id],
+                )
+                .context("backlink chunk to doc-subtree summary")?;
+            } else {
+                tx.execute(
+                    "UPDATE mem_tree_summaries SET parent_id = ?1 \
+                       WHERE id = ?2 AND parent_id IS NULL",
+                    rusqlite::params![&summary_id_for_tx, child_id],
+                )
+                .context("backlink summary to doc-subtree parent")?;
+            }
+        }
+        tx.commit()?;
+        Ok(())
+    })?;
+
+    log::info!(
+        "[tree::bucket_seal] doc-subtree sealed tree_id={} doc_id={:?} level={}→{} summary_id={} children={}",
+        tree.id,
+        doc_id,
+        level,
+        target_level,
+        summary_id,
+        child_ids.len()
+    );
+
+    Ok(node)
+}
+
 #[cfg(test)]
 #[path = "bucket_seal_tests.rs"]
 mod tests;

src/openhuman/memory_tree/tree/bucket_seal_tests.rs

@@ -85,6 +85,272 @@ async fn append_below_budget_does_not_seal() {
     assert_eq!(store::count_summaries(&cfg, &tree.id).unwrap(), 0);
 }
 
+/// Build + persist a Notion-style Document chunk (staged to disk so the
+/// seal hydrator can read its body).
+fn seed_doc_chunk(
+    cfg: &Config,
+    doc_id: &str,
+    seq: u32,
+    content: &str,
+) -> crate::openhuman::memory_store::chunks::types::Chunk {
+    use crate::openhuman::memory_store::chunks::store::upsert_chunks;
+    use crate::openhuman::memory_store::chunks::types::{
+        chunk_id, Chunk, Metadata, SourceKind, SourceRef,
+    };
+    let ts = Utc::now();
+    let c = Chunk {
+        id: chunk_id(SourceKind::Document, doc_id, seq, content),
+        content: content.to_string(),
+        metadata: Metadata {
+            source_kind: SourceKind::Document,
+            source_id: doc_id.to_string(),
+            owner: "notion:conn1".into(),
+            timestamp: ts,
+            time_range: (ts, ts),
+            tags: vec!["notion".into()],
+            source_ref: Some(SourceRef::new("notion://page/pageA")),
+            path_scope: Some("notion:conn1".into()),
+        },
+        token_count: 10,
+        seq_in_source: seq,
+        created_at: ts,
+        partial_message: false,
+    };
+    upsert_chunks(cfg, &[c.clone()]).unwrap();
+    stage_test_chunks(cfg, &[c.clone()]);
+    c
+}
+
+#[tokio::test]
+async fn seal_document_subtree_force_seals_small_doc_to_one_root() {
+    let (_tmp, cfg) = test_config();
+    let tree = get_or_create_source_tree(&cfg, "notion:conn1").unwrap();
+    let provider: Arc<dyn ChatProvider> = Arc::new(StaticChatProvider::new("doc summary"));
+
+    let doc_id = "notion:conn1:pageA";
+    let c0 = seed_doc_chunk(&cfg, doc_id, 0, "first chunk body");
+    let c1 = seed_doc_chunk(&cfg, doc_id, 1, "second chunk body");
+
+    let doc_root_id = test_override::with_provider(provider, async {
+        seal_document_subtree(
+            &cfg,
+            &tree,
+            doc_id,
+            Some(100),
+            &[c0.id.clone(), c1.id.clone()],
+            &LabelStrategy::Empty,
+        )
+        .await
+        .unwrap()
+    })
+    .await;
+
+    // Two small chunks collapse to exactly ONE doc-root, tagged with the
+    // document id + version.
+    let root = store::get_summary(&cfg, &doc_root_id).unwrap().unwrap();
+    assert_eq!(root.doc_id.as_deref(), Some(doc_id));
+    assert_eq!(root.version_ms, Some(100));
+    assert_eq!(
+        root.child_ids.len(),
+        2,
+        "both chunks roll into the doc-root"
+    );
+
+    // The doc-root is fed into the cross-document merge buffer (not the flat
+    // L0 buffer), so the connection tree can fold it with other documents.
+    let merge_buf = store::get_buffer(&cfg, &tree.id, MERGE_LEVEL_BASE).unwrap();
+    assert!(
+        merge_buf.item_ids.contains(&doc_root_id),
+        "doc-root must land in the merge buffer; got {:?}",
+        merge_buf.item_ids
+    );
+    // Per-doc subtree must NOT pollute the flat L0 buffer.
+    let l0 = store::get_buffer(&cfg, &tree.id, 0).unwrap();
+    assert!(
+        l0.item_ids.is_empty(),
+        "L0 buffer stays empty for documents"
+    );
+}
+
+#[tokio::test]
+async fn seal_document_subtree_new_version_is_additive() {
+    let (_tmp, cfg) = test_config();
+    let tree = get_or_create_source_tree(&cfg, "notion:conn1").unwrap();
+    let provider: Arc<dyn ChatProvider> = Arc::new(StaticChatProvider::new("doc summary"));
+
+    let doc_id = "notion:conn1:pageA";
+
+    // Version 1.
+    let v1c = seed_doc_chunk(&cfg, doc_id, 0, "v1 body");
+    let v1_root = test_override::with_provider(Arc::clone(&provider), async {
+        seal_document_subtree(
+            &cfg,
+            &tree,
+            doc_id,
+            Some(100),
+            &[v1c.id.clone()],
+            &LabelStrategy::Empty,
+        )
+        .await
+        .unwrap()
+    })
+    .await;
+
+    // Version 2 (edited page → new chunk content → new chunk id).
+    let v2c = seed_doc_chunk(&cfg, doc_id, 0, "v2 body edited");
+    let v2_root = test_override::with_provider(Arc::clone(&provider), async {
+        seal_document_subtree(
+            &cfg,
+            &tree,
+            doc_id,
+            Some(200),
+            &[v2c.id.clone()],
+            &LabelStrategy::Empty,
+        )
+        .await
+        .unwrap()
+    })
+    .await;
+
+    assert_ne!(v1_root, v2_root, "a new version mints a new doc-root");
+
+    // Forward-only: BOTH doc-roots persist (nothing tombstoned), and both
+    // sit in the merge buffer. Read-time latest-wins (drill_down) is what
+    // surfaces only v2 — the write path never destroys v1.
+    let r1 = store::get_summary(&cfg, &v1_root).unwrap().unwrap();
+    let r2 = store::get_summary(&cfg, &v2_root).unwrap().unwrap();
+    assert_eq!(r1.version_ms, Some(100));
+    assert_eq!(r2.version_ms, Some(200));
+
+    let merge_buf = store::get_buffer(&cfg, &tree.id, MERGE_LEVEL_BASE).unwrap();
+    assert!(merge_buf.item_ids.contains(&v1_root));
+    assert!(merge_buf.item_ids.contains(&v2_root));
+}
+
+/// A byte-identical body chunk reused across two versions of a multi-chunk doc
+/// upserts to the SAME row (content-addressed id). Its single
+/// `parent_summary_id` backlink must follow the NEWEST version's doc-root — the
+/// one drill_down surfaces — not stay stranded on the first (now-superseded)
+/// version. Guards the unconditional re-point in `seal_explicit_children`.
+#[tokio::test]
+async fn shared_chunk_backlink_repoints_to_latest_doc_version() {
+    use crate::openhuman::memory_store::chunks::store::with_connection;
+
+    let (_tmp, cfg) = test_config();
+    let tree = get_or_create_source_tree(&cfg, "notion:conn1").unwrap();
+    let provider: Arc<dyn ChatProvider> = Arc::new(StaticChatProvider::new("doc summary"));
+    let doc_id = "notion:conn1:pageA";
+
+    // seq 0 is byte-identical across versions → one shared row. seq 1 differs,
+    // so each version is a genuine 2-chunk doc (not the single-chunk passthrough).
+    let shared = seed_doc_chunk(&cfg, doc_id, 0, "shared body identical across versions");
+
+    let v1_other = seed_doc_chunk(&cfg, doc_id, 1, "v1 second chunk");
+    let v1_root = test_override::with_provider(Arc::clone(&provider), async {
+        seal_document_subtree(
+            &cfg,
+            &tree,
+            doc_id,
+            Some(100),
+            &[shared.id.clone(), v1_other.id.clone()],
+            &LabelStrategy::Empty,
+        )
+        .await
+        .unwrap()
+    })
+    .await;
+
+    // After v1, the shared chunk backlinks to v1's doc-root.
+    let p1: Option<String> = with_connection(&cfg, |conn| {
+        Ok(conn
+            .query_row(
+                "SELECT parent_summary_id FROM mem_tree_chunks WHERE id = ?1",
+                rusqlite::params![shared.id],
+                |r| r.get(0),
+            )
+            .unwrap())
+    })
+    .unwrap();
+    assert_eq!(p1.as_deref(), Some(v1_root.as_str()));
+
+    // Re-ingest the same shared chunk (idempotent upsert) and seal version 2.
+    let _shared_again = seed_doc_chunk(&cfg, doc_id, 0, "shared body identical across versions");
+    let v2_other = seed_doc_chunk(&cfg, doc_id, 1, "v2 second chunk edited");
+    let v2_root = test_override::with_provider(Arc::clone(&provider), async {
+        seal_document_subtree(
+            &cfg,
+            &tree,
+            doc_id,
+            Some(200),
+            &[shared.id.clone(), v2_other.id.clone()],
+            &LabelStrategy::Empty,
+        )
+        .await
+        .unwrap()
+    })
+    .await;
+    assert_ne!(v1_root, v2_root);
+
+    // The shared chunk's backlink now follows the LATEST version's doc-root.
+    let p2: Option<String> = with_connection(&cfg, |conn| {
+        Ok(conn
+            .query_row(
+                "SELECT parent_summary_id FROM mem_tree_chunks WHERE id = ?1",
+                rusqlite::params![shared.id],
+                |r| r.get(0),
+            )
+            .unwrap())
+    })
+    .unwrap();
+    assert_eq!(
+        p2.as_deref(),
+        Some(v2_root.as_str()),
+        "shared chunk must re-point to the latest doc-root, not stay on v1"
+    );
+}
+
+/// Single-chunk passthrough: a doc that rolls up from exactly one
+/// budget-fitting chunk must NOT invoke the summariser — the doc-root content
+/// is the chunk **verbatim**. Proven two ways: (1) no `ChatProvider` override
+/// is installed, so any summarise() call would hit the (unconfigured) cloud
+/// path and never reproduce this exact text; (2) the doc-root body is asserted
+/// byte-equal to the chunk body.
+#[tokio::test]
+async fn seal_document_subtree_single_chunk_is_verbatim_passthrough_no_llm() {
+    use crate::openhuman::memory_store::content::read as content_read;
+
+    let (_tmp, cfg) = test_config();
+    let tree = get_or_create_source_tree(&cfg, "notion:conn1").unwrap();
+    let doc_id = "notion:conn1:pageX";
+    // Distinctive content the summariser would never emit verbatim.
+    let unique = "UNIQUE-PASSTHROUGH-MARKER-7Z\n\n- line one\n- line two";
+    let c = seed_doc_chunk(&cfg, doc_id, 0, unique);
+
+    // NOTE: no test_override::with_provider — passthrough must not need the LLM.
+    let doc_root_id = seal_document_subtree(
+        &cfg,
+        &tree,
+        doc_id,
+        Some(100),
+        &[c.id.clone()],
+        &LabelStrategy::Empty,
+    )
+    .await
+    .unwrap();
+
+    let root = store::get_summary(&cfg, &doc_root_id).unwrap().unwrap();
+    assert_eq!(root.doc_id.as_deref(), Some(doc_id));
+    assert_eq!(root.version_ms, Some(100));
+
+    // Doc-root body (read full from disk) must be the chunk verbatim.
+    let body = content_read::read_summary_body(&cfg, &doc_root_id).unwrap();
+    assert_eq!(
+        body.trim(),
+        unique,
+        "single-chunk doc-root must be the chunk verbatim (no summarisation / LLM)"
+    );
+}
+
 #[tokio::test]
 async fn crossing_budget_triggers_seal() {
     use crate::openhuman::memory_store::chunks::store::upsert_chunks;
@@ -757,6 +1023,8 @@ async fn hydrate_summary_inputs_batch_preserves_order_and_skips_missing_ids() {
         sealed_at: ts,
         deleted: false,
         embedding: None,
+        doc_id: None,
+        version_ms: None,
     };
     let sum_b = SummaryNode {
         id: "sum-b".into(),
@@ -775,6 +1043,8 @@ async fn hydrate_summary_inputs_batch_preserves_order_and_skips_missing_ids() {
         sealed_at: ts,
         deleted: false,
         embedding: None,
+        doc_id: None,
+        version_ms: None,
     };
 
     // Stage bodies to disk + record content pointers so

src/openhuman/memory_tree/tree/mod.rs

@@ -26,6 +26,9 @@ pub use crate::openhuman::memory_store::trees::{
     Buffer, SummaryNode, Tree, TreeKind, TreeStatus, INPUT_TOKEN_BUDGET, OUTPUT_TOKEN_BUDGET,
     SUMMARY_FANOUT,
 };
-pub use bucket_seal::{append_leaf, append_leaf_deferred, LabelStrategy, LeafRef};
+pub use bucket_seal::{
+    append_leaf, append_leaf_deferred, seal_document_subtree, LabelStrategy, LeafRef,
+    MERGE_LEVEL_BASE,
+};
 pub use factory::{TreeFactory, TreeProfile, GLOBAL_SCOPE};
 pub use registry::{get_or_create_tree, new_summary_id, new_tree_id};

tests/memory_core_threads_raw_coverage_e2e.rs

@@ -183,6 +183,8 @@ fn daily_node(id: &str, tree_id: &str, day: chrono::DateTime<Utc>) -> SummaryNod
         sealed_at: day + Duration::hours(2),
         deleted: false,
         embedding: Some(vec![0.0; 1024]),
+        doc_id: None,
+        version_ms: None,
     }
 }
 
@@ -284,6 +286,8 @@ async fn memory_read_rpc_filters_graphs_scores_reset_and_wipe_seeded_rows() {
         sealed_at: ts0 + Duration::hours(3),
         deleted: false,
         embedding: Some(vec![0.0; 1024]),
+        doc_id: None,
+        version_ms: None,
     };
     insert_summary(
         &cfg,

tests/memory_sync_tree_round21_raw_coverage_e2e.rs

@@ -533,6 +533,8 @@ fn seed_source_summary(
         sealed_at: ts,
         deleted: false,
         embedding: embedding.clone(),
+        doc_id: None,
+        version_ms: None,
     };
     let staged = stage_summary(
         &config.memory_tree_content_root(),

tests/memory_threads_raw_coverage_e2e.rs

@@ -1951,6 +1951,8 @@ fn memory_retrieval_embedding_and_rpc_model_helpers_round_trip() {
         sealed_at: now,
         deleted: false,
         embedding: None,
+        doc_id: None,
+        version_ms: None,
     };
     let tree = Tree {
         id: "tree-1".into(),

tests/memory_tree_sync_deep_raw_coverage_e2e.rs

@@ -194,6 +194,8 @@ fn seed_topic_summary(
         sealed_at: ts,
         deleted: false,
         embedding: None,
+        doc_id: None,
+        version_ms: None,
     };
 
     with_connection(cfg, |conn| {