feat(memory): Phase 1 memory tree - multi-source ingestion & canonical chunks (#707) (#732)

* feat(memory): phase 1 memory tree - multi-source ingestion & canonical chunks (#707)

Adds an isolated memory tree layer under src/openhuman/memory/tree/
implementing Phase 1 of the new memory architecture (umbrella #711).
Zero edits to existing memory/*.rs files - the new layer coexists
with the legacy TinyHumans-backed client.

- Source adapters: chat / email / document -> canonical Markdown
- Token-bounded chunker with deterministic SHA-256 chunk IDs
- SQLite persistence at <workspace>/memory_tree/chunks.db with full
  provenance metadata (source_kind, source_id, owner, timestamps,
  tags, time_range) and back-pointer to raw source
- Unified JSON-RPC ingest (dispatches on source_kind + JSON payload):
  openhuman.memory_tree_ingest, _list_chunks, _get_chunk
- DataSource enum covering the 8 providers from m.excalidraw step 1
  (Discord/Telegram/Whatsapp/Gmail/OtherEmail/Notion/MeetingNotes/DriveDocs)
- ~40 unit tests (chunk ID stability, UTF-8-safe splitting,
  canonicalisation idempotence, store round-trip, filter behavior)

Additive only: new tables in a new DB file, new JSON-RPC namespace,
no existing behavior changes. Feeds #708 (scoring), #709 (summary
trees), #710 (query tools).

Closes #707. Parent: #711.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* feat(memory): enhance memory tree functionality and tests

- Added support for "memory_tree" namespace description in `namespace_description`.
- Implemented clamping for zero token budget in `chunk_markdown` to prevent empty leading chunks.
- Introduced detailed logging for document ingestion, including title length.
- Updated output schemas in `schemas.rs` for improved clarity.
- Enhanced chunk listing with clamping limits and ordering by sequence in `list_chunks`.
- Normalized source references in canonicalization functions to drop blank values.
- Added comprehensive tests for new features and edge cases in chunking and canonicalization.

These changes improve the robustness and usability of the memory tree layer, aligning with ongoing development efforts for Phase 1 of the memory architecture.

* fix(memory): address follow-up CodeRabbit comments

---------

Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Co-authored-by: Steven Enamakel <enamakel@tinyhumans.ai>

src/core/all.rs

@@ -116,6 +116,8 @@ fn build_registered_controllers() -> Vec<RegisteredController> {
     controllers.extend(crate::openhuman::tools::all_tools_registered_controllers());
     // Document and knowledge graph storage
     controllers.extend(crate::openhuman::memory::all_memory_registered_controllers());
+    // Memory tree ingestion layer (#707 — canonicalised chunks with provenance)
+    controllers.extend(crate::openhuman::memory::all_memory_tree_registered_controllers());
     // Referral and growth tracking
     controllers.extend(crate::openhuman::referral::all_referral_registered_controllers());
     // Billing and subscription management
@@ -174,6 +176,7 @@ fn build_declared_controller_schemas() -> Vec<ControllerSchema> {
     schemas.extend(crate::openhuman::workspace::all_workspace_controller_schemas());
     schemas.extend(crate::openhuman::tools::all_tools_controller_schemas());
     schemas.extend(crate::openhuman::memory::all_memory_controller_schemas());
+    schemas.extend(crate::openhuman::memory::all_memory_tree_controller_schemas());
     schemas.extend(crate::openhuman::referral::all_referral_controller_schemas());
     schemas.extend(crate::openhuman::billing::all_billing_controller_schemas());
     schemas.extend(crate::openhuman::team::all_team_controller_schemas());
@@ -230,6 +233,9 @@ pub fn namespace_description(namespace: &str) -> Option<&'static str> {
         "service" => Some("Desktop service lifecycle management."),
         "socket" => Some("Skills runtime socket bridge controls."),
         "memory" => Some("Document storage, vector search, key-value store, and knowledge graph."),
+        "memory_tree" => Some(
+            "Canonical chunk ingestion, provenance capture, and chunk retrieval for source-grounded memory.",
+        ),
         "referral" => Some("Referral codes, stats, and apply flows via the hosted backend API."),
         "billing" => Some("Subscription plan, payment links, and credit top-up via the backend."),
         "team" => Some("Team member management, invites, and role changes via the backend."),
@@ -511,6 +517,7 @@ mod tests {
     #[test]
     fn namespace_description_known_namespaces() {
         assert!(namespace_description("memory").is_some());
+        assert!(namespace_description("memory_tree").is_some());
         assert!(namespace_description("billing").is_some());
         assert!(namespace_description("config").is_some());
         assert!(namespace_description("health").is_some());

src/openhuman/memory/mod.rs

@@ -16,6 +16,7 @@ pub mod rpc_models;
 pub mod schemas;
 pub mod store;
 pub mod traits;
+pub mod tree;
 
 pub use ingestion::{
     ExtractedEntity, ExtractedRelation, ExtractionMode, MemoryIngestionConfig,
@@ -36,3 +37,4 @@ pub use store::{
     NamespaceQueryResult, NamespaceRetrievalContext, RetrievalScoreBreakdown, UnifiedMemory,
 };
 pub use traits::{Memory, MemoryCategory, MemoryEntry};
+pub use tree::{all_memory_tree_controller_schemas, all_memory_tree_registered_controllers};

src/openhuman/memory/tree/canonicalize/chat.rs

@@ -0,0 +1,220 @@
+//! Chat transcripts → canonical Markdown.
+//!
+//! Chat sources are scoped by **channel or group**. A batch of chat messages
+//! from the same channel becomes one [`CanonicalisedSource`]; the chunker
+//! slices it by token budget downstream.
+//!
+//! Output format:
+//! ```md
+//! # Chat transcript — {platform} / {channel}
+//!
+//! ## 2026-04-21T10:12:00Z — Alice
+//! Message body here.
+//!
+//! ## 2026-04-21T10:12:40Z — Bob
+//! Reply body here.
+//! ```
+
+use chrono::{DateTime, Utc};
+use serde::{Deserialize, Serialize};
+
+use super::{normalize_source_ref, CanonicalisedSource};
+use crate::openhuman::memory::tree::types::{Metadata, SourceKind};
+
+/// One chat message in a channel/group.
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub struct ChatMessage {
+    /// Author display name or id.
+    pub author: String,
+    /// When the message was sent.
+    #[serde(with = "chrono::serde::ts_milliseconds")]
+    pub timestamp: DateTime<Utc>,
+    /// Plain text / markdown body.
+    pub text: String,
+    /// Optional per-message provenance pointer (permalink or `platform://...`).
+    #[serde(default)]
+    pub source_ref: Option<String>,
+}
+
+/// Adapter input — a batch of messages from one logical channel.
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub struct ChatBatch {
+    /// Platform name used in the header (e.g. `slack`, `discord`, `telegram`).
+    pub platform: String,
+    /// Human-readable channel / group name for the header.
+    pub channel_label: String,
+    /// Ordered messages (chronological; adapter sorts defensively).
+    pub messages: Vec<ChatMessage>,
+}
+
+/// Canonicalise a chat batch.
+///
+/// Returns `Ok(None)` if the batch has zero messages — callers treat that as
+/// "nothing to ingest" and skip.
+pub fn canonicalise(
+    source_id: &str,
+    owner: &str,
+    tags: &[String],
+    batch: ChatBatch,
+) -> Result<Option<CanonicalisedSource>, String> {
+    if batch.messages.is_empty() {
+        return Ok(None);
+    }
+    let mut messages = batch.messages;
+    messages.sort_by_key(|m| m.timestamp);
+
+    let first_ts = messages.first().map(|m| m.timestamp).unwrap();
+    let last_ts = messages.last().map(|m| m.timestamp).unwrap();
+
+    let mut md = String::new();
+    md.push_str(&format!(
+        "# Chat transcript — {} / {}\n\n",
+        batch.platform, batch.channel_label
+    ));
+    for msg in &messages {
+        md.push_str(&format!(
+            "## {} — {}\n{}\n\n",
+            msg.timestamp.to_rfc3339(),
+            msg.author,
+            msg.text.trim()
+        ));
+    }
+
+    // Provenance points at the batch's first message by default (or whatever
+    // the caller passed on the first message).
+    let source_ref = normalize_source_ref(messages.first().and_then(|m| m.source_ref.clone()));
+
+    let metadata = Metadata {
+        source_kind: SourceKind::Chat,
+        source_id: source_id.to_string(),
+        owner: owner.to_string(),
+        timestamp: first_ts,
+        time_range: (first_ts, last_ts),
+        tags: tags.to_vec(),
+        source_ref,
+    };
+    Ok(Some(CanonicalisedSource {
+        markdown: md,
+        metadata,
+    }))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use chrono::TimeZone;
+
+    fn msg(ts_ms: i64, author: &str, text: &str) -> ChatMessage {
+        ChatMessage {
+            author: author.to_string(),
+            timestamp: Utc.timestamp_millis_opt(ts_ms).unwrap(),
+            text: text.to_string(),
+            source_ref: Some(format!("slack://x/{ts_ms}")),
+        }
+    }
+
+    #[test]
+    fn empty_batch_returns_none() {
+        let b = ChatBatch {
+            platform: "slack".into(),
+            channel_label: "#eng".into(),
+            messages: vec![],
+        };
+        assert!(canonicalise("slack:#eng", "alice", &[], b)
+            .unwrap()
+            .is_none());
+    }
+
+    #[test]
+    fn messages_are_sorted_and_range_captured() {
+        let b = ChatBatch {
+            platform: "slack".into(),
+            channel_label: "#eng".into(),
+            messages: vec![
+                msg(2000, "bob", "second"),
+                msg(1000, "alice", "first"),
+                msg(3000, "carol", "third"),
+            ],
+        };
+        let out = canonicalise("slack:#eng", "alice", &["eng".into()], b)
+            .unwrap()
+            .unwrap();
+        assert_eq!(out.metadata.time_range.0.timestamp_millis(), 1000);
+        assert_eq!(out.metadata.time_range.1.timestamp_millis(), 3000);
+        // Check order in markdown
+        let pos_first = out.markdown.find("first").unwrap();
+        let pos_second = out.markdown.find("second").unwrap();
+        let pos_third = out.markdown.find("third").unwrap();
+        assert!(pos_first < pos_second);
+        assert!(pos_second < pos_third);
+    }
+
+    #[test]
+    fn includes_header_and_per_message_sections() {
+        let b = ChatBatch {
+            platform: "slack".into(),
+            channel_label: "#eng".into(),
+            messages: vec![msg(1000, "alice", "hello")],
+        };
+        let out = canonicalise("slack:#eng", "alice", &[], b)
+            .unwrap()
+            .unwrap();
+        assert!(out
+            .markdown
+            .starts_with("# Chat transcript — slack / #eng\n\n"));
+        assert!(out.markdown.contains("## "));
+        assert!(out.markdown.contains("— alice"));
+        assert!(out.markdown.contains("hello"));
+    }
+
+    #[test]
+    fn source_ref_taken_from_first_message() {
+        let b = ChatBatch {
+            platform: "slack".into(),
+            channel_label: "#eng".into(),
+            messages: vec![msg(1000, "alice", "hi"), msg(2000, "bob", "hey")],
+        };
+        let out = canonicalise("slack:#eng", "alice", &[], b)
+            .unwrap()
+            .unwrap();
+        assert_eq!(
+            out.metadata.source_ref.as_ref().unwrap().value,
+            "slack://x/1000"
+        );
+    }
+
+    #[test]
+    fn metadata_carries_owner_and_tags() {
+        let b = ChatBatch {
+            platform: "slack".into(),
+            channel_label: "#eng".into(),
+            messages: vec![msg(1000, "alice", "hi")],
+        };
+        let out = canonicalise(
+            "slack:#eng",
+            "alice@example.com",
+            &["eng".into(), "on-call".into()],
+            b,
+        )
+        .unwrap()
+        .unwrap();
+        assert_eq!(out.metadata.owner, "alice@example.com");
+        assert_eq!(out.metadata.tags, vec!["eng", "on-call"]);
+        assert_eq!(out.metadata.source_kind, SourceKind::Chat);
+    }
+
+    #[test]
+    fn blank_source_ref_is_dropped() {
+        let mut first = msg(1000, "alice", "hi");
+        first.source_ref = Some("   ".into());
+        let b = ChatBatch {
+            platform: "slack".into(),
+            channel_label: "#eng".into(),
+            messages: vec![first],
+        };
+        let out = canonicalise("slack:#eng", "alice", &[], b)
+            .unwrap()
+            .unwrap();
+        assert!(out.metadata.source_ref.is_none());
+    }
+}

src/openhuman/memory/tree/canonicalize/document.rs

@@ -0,0 +1,130 @@
+//! Standalone documents → canonical Markdown.
+//!
+//! Document sources are single-record (no grouping): one Notion page, one
+//! Drive doc, one meeting-note file. The canonicaliser adds a small title
+//! header and passes through the body; if the body is already markdown it
+//! is kept verbatim.
+
+use chrono::{DateTime, Utc};
+use serde::{Deserialize, Serialize};
+
+use super::{normalize_source_ref, CanonicalisedSource};
+use crate::openhuman::memory::tree::types::{Metadata, SourceKind};
+
+/// Adapter input for a single document.
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub struct DocumentInput {
+    /// Provider name (e.g. `notion`, `drive`, `meeting_notes`).
+    pub provider: String,
+    /// Document title.
+    pub title: String,
+    /// Document body (markdown preferred; plain text also accepted).
+    pub body: String,
+    /// When the document was last modified at the source.
+    #[serde(with = "chrono::serde::ts_milliseconds")]
+    pub modified_at: DateTime<Utc>,
+    /// Optional pointer back to source (URL, file path, Notion page id).
+    #[serde(default)]
+    pub source_ref: Option<String>,
+}
+
+pub fn canonicalise(
+    source_id: &str,
+    owner: &str,
+    tags: &[String],
+    doc: DocumentInput,
+) -> Result<Option<CanonicalisedSource>, String> {
+    if doc.body.trim().is_empty() && doc.title.trim().is_empty() {
+        return Ok(None);
+    }
+
+    let mut md = String::new();
+    md.push_str(&format!("# {} — {}\n\n", doc.provider, doc.title));
+    md.push_str(doc.body.trim());
+    md.push('\n');
+
+    Ok(Some(CanonicalisedSource {
+        markdown: md,
+        metadata: Metadata {
+            source_kind: SourceKind::Document,
+            source_id: source_id.to_string(),
+            owner: owner.to_string(),
+            timestamp: doc.modified_at,
+            time_range: (doc.modified_at, doc.modified_at),
+            tags: tags.to_vec(),
+            source_ref: normalize_source_ref(doc.source_ref),
+        },
+    }))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use chrono::TimeZone;
+
+    fn doc(title: &str, body: &str) -> DocumentInput {
+        DocumentInput {
+            provider: "notion".into(),
+            title: title.into(),
+            body: body.into(),
+            modified_at: Utc.timestamp_millis_opt(1_700_000_000_000).unwrap(),
+            source_ref: Some("notion://page/abc".into()),
+        }
+    }
+
+    #[test]
+    fn empty_doc_returns_none() {
+        let d = DocumentInput {
+            provider: "notion".into(),
+            title: "".into(),
+            body: "   \n  ".into(),
+            modified_at: Utc::now(),
+            source_ref: None,
+        };
+        assert!(canonicalise("d1", "alice", &[], d).unwrap().is_none());
+    }
+
+    #[test]
+    fn renders_title_and_body() {
+        let out = canonicalise(
+            "d1",
+            "alice",
+            &[],
+            doc("Launch plan", "step one\n\nstep two"),
+        )
+        .unwrap()
+        .unwrap();
+        assert!(out.markdown.starts_with("# notion — Launch plan\n\n"));
+        assert!(out.markdown.contains("step one"));
+        assert!(out.markdown.contains("step two"));
+    }
+
+    #[test]
+    fn metadata_single_point_time_range() {
+        let out = canonicalise("d1", "alice", &[], doc("x", "y"))
+            .unwrap()
+            .unwrap();
+        assert_eq!(out.metadata.time_range.0, out.metadata.time_range.1);
+        assert_eq!(out.metadata.source_kind, SourceKind::Document);
+    }
+
+    #[test]
+    fn source_ref_carried_through() {
+        let out = canonicalise("d1", "alice", &["proj".into()], doc("x", "y"))
+            .unwrap()
+            .unwrap();
+        assert_eq!(
+            out.metadata.source_ref.as_ref().unwrap().value,
+            "notion://page/abc"
+        );
+        assert_eq!(out.metadata.tags, vec!["proj"]);
+    }
+
+    #[test]
+    fn blank_source_ref_is_dropped() {
+        let mut input = doc("x", "y");
+        input.source_ref = Some(" \n ".into());
+        let out = canonicalise("d1", "alice", &[], input).unwrap().unwrap();
+        assert!(out.metadata.source_ref.is_none());
+    }
+}

src/openhuman/memory/tree/canonicalize/email.rs

@@ -0,0 +1,184 @@
+//! Email threads → canonical Markdown.
+//!
+//! Email sources are scoped by **thread**. One thread becomes one
+//! [`CanonicalisedSource`]. Headers (From, To, Subject, Date) surface in a
+//! small frontmatter-style block per message; the body follows as markdown.
+
+use chrono::{DateTime, Utc};
+use serde::{Deserialize, Serialize};
+
+use super::{normalize_source_ref, CanonicalisedSource};
+use crate::openhuman::memory::tree::types::{Metadata, SourceKind};
+
+/// One email in a thread.
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub struct EmailMessage {
+    pub from: String,
+    #[serde(default)]
+    pub to: Vec<String>,
+    #[serde(default)]
+    pub cc: Vec<String>,
+    pub subject: String,
+    #[serde(with = "chrono::serde::ts_milliseconds")]
+    pub sent_at: DateTime<Utc>,
+    /// Plain-text or markdown body.
+    pub body: String,
+    /// Message-id header or provider URL; used for citation back to source.
+    #[serde(default)]
+    pub source_ref: Option<String>,
+}
+
+/// A whole email thread.
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub struct EmailThread {
+    /// Provider name used in the header (e.g. `gmail`, `outlook`).
+    pub provider: String,
+    /// Thread subject shown on top (usually the subject of the first message).
+    pub thread_subject: String,
+    /// Ordered messages (chronological; adapter sorts defensively).
+    pub messages: Vec<EmailMessage>,
+}
+
+pub fn canonicalise(
+    source_id: &str,
+    owner: &str,
+    tags: &[String],
+    thread: EmailThread,
+) -> Result<Option<CanonicalisedSource>, String> {
+    if thread.messages.is_empty() {
+        return Ok(None);
+    }
+    let mut messages = thread.messages;
+    messages.sort_by_key(|m| m.sent_at);
+
+    let first_ts = messages.first().map(|m| m.sent_at).unwrap();
+    let last_ts = messages.last().map(|m| m.sent_at).unwrap();
+
+    let mut md = String::new();
+    md.push_str(&format!(
+        "# Email thread — {} — {}\n\n",
+        thread.provider, thread.thread_subject
+    ));
+
+    for msg in &messages {
+        md.push_str("---\n");
+        md.push_str(&format!("From: {}\n", msg.from));
+        if !msg.to.is_empty() {
+            md.push_str(&format!("To: {}\n", msg.to.join(", ")));
+        }
+        if !msg.cc.is_empty() {
+            md.push_str(&format!("Cc: {}\n", msg.cc.join(", ")));
+        }
+        md.push_str(&format!("Subject: {}\n", msg.subject));
+        md.push_str(&format!("Date: {}\n\n", msg.sent_at.to_rfc3339()));
+        md.push_str(msg.body.trim());
+        md.push_str("\n\n");
+    }
+
+    let source_ref = normalize_source_ref(messages.first().and_then(|m| m.source_ref.clone()));
+
+    Ok(Some(CanonicalisedSource {
+        markdown: md,
+        metadata: Metadata {
+            source_kind: SourceKind::Email,
+            source_id: source_id.to_string(),
+            owner: owner.to_string(),
+            timestamp: first_ts,
+            time_range: (first_ts, last_ts),
+            tags: tags.to_vec(),
+            source_ref,
+        },
+    }))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use chrono::TimeZone;
+
+    fn email(ts_ms: i64, from: &str, subject: &str, body: &str) -> EmailMessage {
+        EmailMessage {
+            from: from.to_string(),
+            to: vec!["alice@example.com".into()],
+            cc: vec![],
+            subject: subject.to_string(),
+            sent_at: Utc.timestamp_millis_opt(ts_ms).unwrap(),
+            body: body.to_string(),
+            source_ref: Some(format!("<msg-{ts_ms}@example.com>")),
+        }
+    }
+
+    #[test]
+    fn empty_thread_returns_none() {
+        let t = EmailThread {
+            provider: "gmail".into(),
+            thread_subject: "x".into(),
+            messages: vec![],
+        };
+        assert!(canonicalise("gmail:t1", "alice", &[], t).unwrap().is_none());
+    }
+
+    #[test]
+    fn renders_headers_and_body_per_message() {
+        let t = EmailThread {
+            provider: "gmail".into(),
+            thread_subject: "Launch".into(),
+            messages: vec![
+                email(1000, "bob@example.com", "Launch", "let's ship"),
+                email(2000, "alice@example.com", "Re: Launch", "agreed"),
+            ],
+        };
+        let out = canonicalise("gmail:t1", "alice@example.com", &[], t)
+            .unwrap()
+            .unwrap();
+        assert!(out.markdown.contains("# Email thread — gmail — Launch"));
+        assert!(out.markdown.contains("From: bob@example.com"));
+        assert!(out.markdown.contains("Subject: Launch"));
+        assert!(out.markdown.contains("let's ship"));
+        assert!(out.markdown.contains("Re: Launch"));
+        assert!(out.markdown.contains("agreed"));
+    }
+
+    #[test]
+    fn time_range_spans_thread() {
+        let t = EmailThread {
+            provider: "gmail".into(),
+            thread_subject: "x".into(),
+            messages: vec![
+                email(3000, "c", "y", "third"),
+                email(1000, "a", "y", "first"),
+                email(2000, "b", "y", "second"),
+            ],
+        };
+        let out = canonicalise("gmail:t1", "a", &[], t).unwrap().unwrap();
+        assert_eq!(out.metadata.time_range.0.timestamp_millis(), 1000);
+        assert_eq!(out.metadata.time_range.1.timestamp_millis(), 3000);
+    }
+
+    #[test]
+    fn source_ref_from_first_message() {
+        let t = EmailThread {
+            provider: "gmail".into(),
+            thread_subject: "x".into(),
+            messages: vec![email(1000, "a", "y", "b"), email(2000, "b", "y", "c")],
+        };
+        let out = canonicalise("gmail:t1", "a", &[], t).unwrap().unwrap();
+        assert_eq!(
+            out.metadata.source_ref.as_ref().unwrap().value,
+            "<msg-1000@example.com>"
+        );
+    }
+
+    #[test]
+    fn blank_source_ref_is_dropped() {
+        let mut first = email(1000, "a", "y", "b");
+        first.source_ref = Some("".into());
+        let t = EmailThread {
+            provider: "gmail".into(),
+            thread_subject: "x".into(),
+            messages: vec![first],
+        };
+        let out = canonicalise("gmail:t1", "a", &[], t).unwrap().unwrap();
+        assert!(out.metadata.source_ref.is_none());
+    }
+}

src/openhuman/memory/tree/canonicalize/mod.rs

@@ -0,0 +1,56 @@
+//! Canonicalisers — normalise source-specific payloads into canonical
+//! Markdown with provenance metadata (Phase 1 / #707).
+//!
+//! Each source kind has its own adapter. They all return the same shape:
+//! a [`CanonicalisedSource`] containing the markdown blob plus a seed
+//! [`Metadata`] that the chunker will clone onto each produced chunk.
+//!
+//! Adapters do not interpret content semantically — they only normalise
+//! shape and capture provenance. Scoring / entity extraction / summarisation
+//! happen downstream in later phases.
+
+pub mod chat;
+pub mod document;
+pub mod email;
+
+use serde::{Deserialize, Serialize};
+
+use crate::openhuman::memory::tree::types::{Metadata, SourceRef};
+
+/// Output of a canonicaliser — one per logical source record
+/// (a chat batch, an email, a document).
+#[derive(Clone, Debug)]
+pub struct CanonicalisedSource {
+    pub markdown: String,
+    pub metadata: Metadata,
+}
+
+/// Shared input shape: a payload + a minimal provenance hint.
+///
+/// Every adapter accepts this generic envelope; the concrete payload type
+/// is adapter-specific (see sibling modules for the per-kind inputs).
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub struct CanonicaliseRequest<P> {
+    /// Logical source id (channel for chat, thread for email, doc id).
+    pub source_id: String,
+    /// Owner / user account.
+    #[serde(default)]
+    pub owner: String,
+    /// Source-specific payload.
+    pub payload: P,
+    /// Optional tags carried through.
+    #[serde(default)]
+    pub tags: Vec<String>,
+}
+
+/// Trim provider-specific source references and drop blank pointers.
+pub fn normalize_source_ref(source_ref: Option<String>) -> Option<SourceRef> {
+    source_ref.and_then(|value| {
+        let trimmed = value.trim();
+        if trimmed.is_empty() {
+            None
+        } else {
+            Some(SourceRef::new(trimmed.to_string()))
+        }
+    })
+}

src/openhuman/memory/tree/chunker.rs

@@ -0,0 +1,347 @@
+//! Markdown → bounded chunks with stable sequence numbers (Phase 1 / #707).
+//!
+//! The canonicalisers produce one big canonical Markdown blob per source
+//! record; the chunker slices that into chunks of at most [`DEFAULT_CHUNK_MAX_TOKENS`]
+//! so later phases (#709 seal budget = 10k tokens) can ingest them without
+//! blowing past the summariser ceiling.
+//!
+//! Splitting strategy: prefer paragraph boundaries (`\n\n`), then single
+//! newlines, then hard character-count cut. We never split a character in half.
+
+use crate::openhuman::memory::tree::types::{approx_token_count, Chunk, Metadata, SourceKind};
+
+/// Default upper bound on per-chunk tokens.
+///
+/// Aligned with the LLD's summariser 10k budget so a single chunk never blows
+/// a seal on its own.
+pub const DEFAULT_CHUNK_MAX_TOKENS: u32 = 10_000;
+
+/// Tunable settings for the chunker.
+#[derive(Clone, Debug)]
+pub struct ChunkerOptions {
+    pub max_tokens: u32,
+}
+
+impl Default for ChunkerOptions {
+    fn default() -> Self {
+        Self {
+            max_tokens: DEFAULT_CHUNK_MAX_TOKENS,
+        }
+    }
+}
+
+/// Input to the chunker: the canonicalised source and its provenance.
+///
+/// Callers (typically canonicalisers via [`super::ingest`]) own construction;
+/// the chunker does not interpret metadata beyond cloning it onto each chunk.
+#[derive(Clone, Debug)]
+pub struct ChunkerInput {
+    pub source_kind: SourceKind,
+    pub source_id: String,
+    /// Canonical Markdown content — possibly very long.
+    pub markdown: String,
+    /// Base metadata; per-chunk `timestamp` defaults to `metadata.timestamp`.
+    pub metadata: Metadata,
+}
+
+/// Slice `input.markdown` into chunks ≤ `opts.max_tokens` tokens each.
+///
+/// Returns chunks in source order with stable sequence numbers starting at 0.
+/// Chunk IDs are deterministic (`types::chunk_id`), so re-chunking yields the
+/// same ids for identical input.
+pub fn chunk_markdown(input: &ChunkerInput, opts: &ChunkerOptions) -> Vec<Chunk> {
+    let now = chrono::Utc::now();
+    let pieces = split_by_token_budget(&input.markdown, opts.max_tokens);
+    log::debug!(
+        "[memory_tree::chunker] source_kind={} source_id={} len={} pieces={}",
+        input.source_kind.as_str(),
+        input.source_id,
+        input.markdown.len(),
+        pieces.len()
+    );
+
+    pieces
+        .into_iter()
+        .enumerate()
+        .map(|(idx, content)| {
+            let seq = idx as u32;
+            let token_count = approx_token_count(&content);
+            Chunk {
+                id: super::types::chunk_id(input.source_kind, &input.source_id, seq),
+                content,
+                metadata: input.metadata.clone(),
+                token_count,
+                seq_in_source: seq,
+                created_at: now,
+            }
+        })
+        .collect()
+}
+
+/// Split `text` into pieces each ≤ `max_tokens` tokens.
+///
+/// Preference order for split boundaries:
+/// 1. Paragraph (`\n\n`)
+/// 2. Line (`\n`)
+/// 3. Hard character cut (last resort; preserves UTF-8 code points)
+fn split_by_token_budget(text: &str, max_tokens: u32) -> Vec<String> {
+    let max_tokens = max_tokens.max(1);
+    if text.is_empty() {
+        return vec![String::new()];
+    }
+    if approx_token_count(text) <= max_tokens {
+        return vec![text.to_string()];
+    }
+
+    // Approximate max chars per chunk (4 chars ≈ 1 token).
+    let max_chars: usize = (max_tokens as usize).saturating_mul(4);
+
+    // First: try paragraph split. Walk paragraphs, greedy-accumulate into
+    // chunks ≤ max_chars.
+    let paragraphs: Vec<&str> = text.split("\n\n").collect();
+    if paragraphs.len() > 1 {
+        if let Some(out) = pack_segments(&paragraphs, "\n\n", max_chars) {
+            return out;
+        }
+    }
+
+    // Fall back to line split.
+    let lines: Vec<&str> = text.split('\n').collect();
+    if lines.len() > 1 {
+        if let Some(out) = pack_segments(&lines, "\n", max_chars) {
+            return out;
+        }
+    }
+
+    // Fall back to hard character-count cut preserving UTF-8 boundaries.
+    hard_split_by_chars(text, max_chars)
+}
+
+/// Greedily pack pre-split segments into chunks ≤ max_chars. Returns `None`
+/// if any single segment is already too large — caller should try a finer
+/// split.
+fn pack_segments(segments: &[&str], sep: &str, max_chars: usize) -> Option<Vec<String>> {
+    let sep_len = sep.len();
+    let mut out: Vec<String> = Vec::new();
+    let mut current = String::new();
+
+    for seg in segments {
+        let seg_len = seg.chars().count();
+        // A single segment larger than max_chars forces a finer split.
+        if seg_len > max_chars {
+            return None;
+        }
+        let projected = if current.is_empty() {
+            seg_len
+        } else {
+            current.chars().count() + sep_len + seg_len
+        };
+        if projected > max_chars {
+            out.push(std::mem::take(&mut current));
+            current.push_str(seg);
+        } else {
+            if !current.is_empty() {
+                current.push_str(sep);
+            }
+            current.push_str(seg);
+        }
+    }
+    if !current.is_empty() {
+        out.push(current);
+    }
+    if out.is_empty() {
+        out.push(String::new());
+    }
+    Some(out)
+}
+
+/// Hard character-count cut preserving UTF-8 code-point boundaries.
+fn hard_split_by_chars(text: &str, max_chars: usize) -> Vec<String> {
+    let mut out = Vec::new();
+    let mut current = String::new();
+    let mut count = 0usize;
+    for ch in text.chars() {
+        if count + 1 > max_chars {
+            out.push(std::mem::take(&mut current));
+            count = 0;
+        }
+        current.push(ch);
+        count += 1;
+    }
+    if !current.is_empty() {
+        out.push(current);
+    }
+    if out.is_empty() {
+        out.push(String::new());
+    }
+    out
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use chrono::Utc;
+
+    fn meta() -> Metadata {
+        Metadata::point_in_time(SourceKind::Chat, "slack:#eng", "alice", Utc::now())
+    }
+
+    #[test]
+    fn tiny_input_produces_single_chunk() {
+        let input = ChunkerInput {
+            source_kind: SourceKind::Chat,
+            source_id: "slack:#eng".into(),
+            markdown: "hello world".into(),
+            metadata: meta(),
+        };
+        let chunks = chunk_markdown(&input, &ChunkerOptions::default());
+        assert_eq!(chunks.len(), 1);
+        assert_eq!(chunks[0].content, "hello world");
+        assert_eq!(chunks[0].seq_in_source, 0);
+    }
+
+    #[test]
+    fn empty_input_produces_one_empty_chunk() {
+        let input = ChunkerInput {
+            source_kind: SourceKind::Chat,
+            source_id: "x".into(),
+            markdown: "".into(),
+            metadata: meta(),
+        };
+        let chunks = chunk_markdown(&input, &ChunkerOptions::default());
+        assert_eq!(chunks.len(), 1);
+        assert_eq!(chunks[0].content, "");
+    }
+
+    #[test]
+    fn paragraph_boundaries_preferred() {
+        // Build something that exceeds token budget so it must split.
+        let para1 = "a".repeat(400); // ~100 tokens
+        let para2 = "b".repeat(400);
+        let para3 = "c".repeat(400);
+        let text = format!("{para1}\n\n{para2}\n\n{para3}");
+        let input = ChunkerInput {
+            source_kind: SourceKind::Document,
+            source_id: "doc1".into(),
+            markdown: text,
+            metadata: meta(),
+        };
+        let chunks = chunk_markdown(
+            &input,
+            &ChunkerOptions {
+                max_tokens: 150, // forces split at paragraph
+            },
+        );
+        assert!(chunks.len() >= 2);
+        // Every chunk should be a multiple of complete paragraphs — so starts
+        // with 'a', 'b', or 'c' and doesn't mix them arbitrarily.
+        for c in &chunks {
+            let first = c.content.chars().next().unwrap();
+            assert!(
+                matches!(first, 'a' | 'b' | 'c'),
+                "chunk starts with unexpected char: {:?}",
+                c.content.chars().take(10).collect::<String>()
+            );
+        }
+    }
+
+    #[test]
+    fn falls_back_to_line_split_when_no_paragraphs() {
+        let text = (0..30)
+            .map(|i| format!("line-{i}-{}", "x".repeat(40)))
+            .collect::<Vec<_>>()
+            .join("\n");
+        let input = ChunkerInput {
+            source_kind: SourceKind::Chat,
+            source_id: "x".into(),
+            markdown: text,
+            metadata: meta(),
+        };
+        let chunks = chunk_markdown(
+            &input,
+            &ChunkerOptions {
+                max_tokens: 80, // forces several splits
+            },
+        );
+        assert!(chunks.len() >= 2);
+        for c in &chunks {
+            assert!(!c.content.contains("\n\n")); // no paragraph joins in output
+        }
+    }
+
+    #[test]
+    fn chunk_ids_are_stable_across_runs() {
+        let input = ChunkerInput {
+            source_kind: SourceKind::Chat,
+            source_id: "slack:#eng".into(),
+            markdown: "hello".into(),
+            metadata: meta(),
+        };
+        let a = chunk_markdown(&input, &ChunkerOptions::default());
+        let b = chunk_markdown(&input, &ChunkerOptions::default());
+        assert_eq!(
+            a.iter().map(|c| c.id.clone()).collect::<Vec<_>>(),
+            b.iter().map(|c| c.id.clone()).collect::<Vec<_>>()
+        );
+    }
+
+    #[test]
+    fn sequence_numbers_start_at_zero() {
+        let text = (0..10)
+            .map(|i| format!("p{i}"))
+            .collect::<Vec<_>>()
+            .join("\n\n");
+        let input = ChunkerInput {
+            source_kind: SourceKind::Document,
+            source_id: "d".into(),
+            markdown: text,
+            metadata: meta(),
+        };
+        let chunks = chunk_markdown(
+            &input,
+            &ChunkerOptions {
+                max_tokens: 2, // forces one chunk per paragraph basically
+            },
+        );
+        for (idx, c) in chunks.iter().enumerate() {
+            assert_eq!(c.seq_in_source, idx as u32);
+        }
+    }
+
+    #[test]
+    fn utf8_boundaries_preserved_on_hard_split() {
+        // Single long line with no paragraph/line splits → falls to hard cut.
+        // Use a 3-byte-per-char Unicode codepoint (Chinese) and force a cut.
+        let text = "中".repeat(400);
+        let input = ChunkerInput {
+            source_kind: SourceKind::Document,
+            source_id: "d".into(),
+            markdown: text.clone(),
+            metadata: meta(),
+        };
+        let chunks = chunk_markdown(
+            &input,
+            &ChunkerOptions {
+                max_tokens: 50, // ~200 chars
+            },
+        );
+        // Rejoining must equal the original.
+        let rejoined: String = chunks.iter().map(|c| c.content.as_str()).collect();
+        assert_eq!(rejoined, text);
+    }
+
+    #[test]
+    fn zero_token_budget_is_clamped_without_empty_leading_chunk() {
+        let input = ChunkerInput {
+            source_kind: SourceKind::Document,
+            source_id: "d".into(),
+            markdown: "abcdef".into(),
+            metadata: meta(),
+        };
+        let chunks = chunk_markdown(&input, &ChunkerOptions { max_tokens: 0 });
+        assert!(!chunks.is_empty());
+        assert!(chunks.iter().all(|chunk| !chunk.content.is_empty()));
+        let rejoined: String = chunks.iter().map(|c| c.content.as_str()).collect();
+        assert_eq!(rejoined, "abcdef");
+    }
+}

src/openhuman/memory/tree/ingest.rs

@@ -0,0 +1,235 @@
+//! Ingest orchestrator: canonicalise → chunk → persist (Phase 1 / #707).
+//!
+//! Consumers call one `ingest_*` function per source kind. Each returns an
+//! [`IngestResult`] so the RPC layer can report how many chunks landed.
+
+use anyhow::Result;
+use serde::{Deserialize, Serialize};
+
+use crate::openhuman::config::Config;
+use crate::openhuman::memory::tree::canonicalize::{
+    chat::{self, ChatBatch},
+    document::{self, DocumentInput},
+    email::{self, EmailThread},
+};
+use crate::openhuman::memory::tree::chunker::{chunk_markdown, ChunkerInput, ChunkerOptions};
+use crate::openhuman::memory::tree::store;
+use crate::openhuman::memory::tree::types::Chunk;
+
+/// Outcome of one ingest call.
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub struct IngestResult {
+    pub source_id: String,
+    pub chunks_written: usize,
+    pub chunk_ids: Vec<String>,
+}
+
+impl IngestResult {
+    fn from_chunks(source_id: String, chunks: &[Chunk], chunks_written: usize) -> Self {
+        Self {
+            source_id,
+            chunks_written,
+            chunk_ids: chunks.iter().map(|c| c.id.clone()).collect(),
+        }
+    }
+}
+
+/// Ingest a batch of chat messages scoped to one channel/group.
+pub fn ingest_chat(
+    config: &Config,
+    source_id: &str,
+    owner: &str,
+    tags: Vec<String>,
+    batch: ChatBatch,
+) -> Result<IngestResult> {
+    log::debug!(
+        "[memory_tree::ingest] chat source_id={} msg_count={}",
+        source_id,
+        batch.messages.len()
+    );
+    let canonical =
+        match chat::canonicalise(source_id, owner, &tags, batch).map_err(anyhow::Error::msg)? {
+            Some(c) => c,
+            None => {
+                return Ok(IngestResult {
+                    source_id: source_id.to_string(),
+                    chunks_written: 0,
+                    chunk_ids: Vec::new(),
+                });
+            }
+        };
+    persist(config, source_id, canonical)
+}
+
+/// Ingest a single email thread.
+pub fn ingest_email(
+    config: &Config,
+    source_id: &str,
+    owner: &str,
+    tags: Vec<String>,
+    thread: EmailThread,
+) -> Result<IngestResult> {
+    log::debug!(
+        "[memory_tree::ingest] email source_id={} msg_count={}",
+        source_id,
+        thread.messages.len()
+    );
+    let canonical =
+        match email::canonicalise(source_id, owner, &tags, thread).map_err(anyhow::Error::msg)? {
+            Some(c) => c,
+            None => {
+                return Ok(IngestResult {
+                    source_id: source_id.to_string(),
+                    chunks_written: 0,
+                    chunk_ids: Vec::new(),
+                });
+            }
+        };
+    persist(config, source_id, canonical)
+}
+
+/// Ingest a single standalone document.
+pub fn ingest_document(
+    config: &Config,
+    source_id: &str,
+    owner: &str,
+    tags: Vec<String>,
+    doc: DocumentInput,
+) -> Result<IngestResult> {
+    let title_len = doc.title.chars().count();
+    log::debug!(
+        "[memory_tree::ingest] document source_id={} has_title={} title_len={}",
+        source_id,
+        !doc.title.trim().is_empty(),
+        title_len
+    );
+    let canonical =
+        match document::canonicalise(source_id, owner, &tags, doc).map_err(anyhow::Error::msg)? {
+            Some(c) => c,
+            None => {
+                return Ok(IngestResult {
+                    source_id: source_id.to_string(),
+                    chunks_written: 0,
+                    chunk_ids: Vec::new(),
+                });
+            }
+        };
+    persist(config, source_id, canonical)
+}
+
+fn persist(
+    config: &Config,
+    source_id: &str,
+    canonical: crate::openhuman::memory::tree::canonicalize::CanonicalisedSource,
+) -> Result<IngestResult> {
+    let input = ChunkerInput {
+        source_kind: canonical.metadata.source_kind,
+        source_id: source_id.to_string(),
+        markdown: canonical.markdown,
+        metadata: canonical.metadata,
+    };
+    let chunks = chunk_markdown(&input, &ChunkerOptions::default());
+    let written = store::upsert_chunks(config, &chunks)?;
+    log::debug!(
+        "[memory_tree::ingest] persisted source_id={} chunks={}",
+        source_id,
+        written
+    );
+    Ok(IngestResult::from_chunks(
+        source_id.to_string(),
+        &chunks,
+        written,
+    ))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::openhuman::memory::tree::canonicalize::chat::ChatMessage;
+    use crate::openhuman::memory::tree::store::{count_chunks, list_chunks, ListChunksQuery};
+    use crate::openhuman::memory::tree::types::SourceKind;
+    use chrono::{TimeZone, Utc};
+    use tempfile::TempDir;
+
+    fn test_config() -> (TempDir, Config) {
+        let tmp = TempDir::new().unwrap();
+        let mut cfg = Config::default();
+        cfg.workspace_dir = tmp.path().to_path_buf();
+        (tmp, cfg)
+    }
+
+    #[test]
+    fn ingest_chat_writes_chunks() {
+        let (_tmp, cfg) = test_config();
+        let batch = ChatBatch {
+            platform: "slack".into(),
+            channel_label: "#eng".into(),
+            messages: vec![
+                ChatMessage {
+                    author: "alice".into(),
+                    timestamp: Utc.timestamp_millis_opt(1_700_000_000_000).unwrap(),
+                    text: "hello".into(),
+                    source_ref: Some("slack://m1".into()),
+                },
+                ChatMessage {
+                    author: "bob".into(),
+                    timestamp: Utc.timestamp_millis_opt(1_700_000_010_000).unwrap(),
+                    text: "world".into(),
+                    source_ref: None,
+                },
+            ],
+        };
+        let out = ingest_chat(&cfg, "slack:#eng", "alice", vec![], batch).unwrap();
+        assert_eq!(out.chunks_written, 1);
+        assert_eq!(count_chunks(&cfg).unwrap(), 1);
+        let rows = list_chunks(&cfg, &ListChunksQuery::default()).unwrap();
+        assert_eq!(rows[0].metadata.source_kind, SourceKind::Chat);
+        assert_eq!(rows[0].metadata.source_id, "slack:#eng");
+    }
+
+    #[test]
+    fn ingest_chat_empty_batch_is_noop() {
+        let (_tmp, cfg) = test_config();
+        let batch = ChatBatch {
+            platform: "slack".into(),
+            channel_label: "#eng".into(),
+            messages: vec![],
+        };
+        let out = ingest_chat(&cfg, "slack:#eng", "alice", vec![], batch).unwrap();
+        assert_eq!(out.chunks_written, 0);
+        assert_eq!(count_chunks(&cfg).unwrap(), 0);
+    }
+
+    #[test]
+    fn re_ingest_is_idempotent() {
+        let (_tmp, cfg) = test_config();
+        let doc = DocumentInput {
+            provider: "notion".into(),
+            title: "Launch plan".into(),
+            body: "content here".into(),
+            modified_at: Utc.timestamp_millis_opt(1_700_000_000_000).unwrap(),
+            source_ref: Some("notion://page/abc".into()),
+        };
+        ingest_document(&cfg, "notion:abc", "alice", vec![], doc.clone()).unwrap();
+        ingest_document(&cfg, "notion:abc", "alice", vec![], doc).unwrap();
+        assert_eq!(count_chunks(&cfg).unwrap(), 1);
+    }
+
+    #[test]
+    fn chunks_preserve_source_ref() {
+        let (_tmp, cfg) = test_config();
+        let doc = DocumentInput {
+            provider: "notion".into(),
+            title: "t".into(),
+            body: "b".into(),
+            modified_at: Utc::now(),
+            source_ref: Some("notion://x".into()),
+        };
+        ingest_document(&cfg, "notion:x", "alice", vec![], doc).unwrap();
+        let rows = list_chunks(&cfg, &ListChunksQuery::default()).unwrap();
+        assert_eq!(
+            rows[0].metadata.source_ref.as_ref().unwrap().value,
+            "notion://x"
+        );
+    }
+}

src/openhuman/memory/tree/mod.rs

@@ -0,0 +1,36 @@
+//! Memory tree ingestion layer (Phase 1 / issue #707).
+//!
+//! This is an isolated subdir under `openhuman::memory` implementing the
+//! new bucket-seal-ready local memory architecture described in
+//! `docs/MEMORY_ARCHITECTURE_LLD.md`. It does **not** share files with the
+//! legacy `memory` module; they coexist until the legacy remote-client
+//! layer is replaced in a future phase.
+//!
+//! Phase 1 scope (this module):
+//! - source adapters (chat / email / document) → canonical Markdown
+//! - chunker with stable deterministic IDs and bounded segments
+//! - SQLite persistence with provenance metadata + back-pointer to raw
+//! - JSON-RPC controllers under the `memory_tree` namespace
+//!
+//! Public RPC surface (see `schemas.rs`):
+//! - `openhuman.memory_tree_ingest` — unified ingest; caller supplies
+//!   `source_kind` (chat|email|document) and a JSON `payload` whose shape
+//!   the handler validates based on the kind
+//! - `openhuman.memory_tree_list_chunks`
+//! - `openhuman.memory_tree_get_chunk`
+//!
+//! Phases 2-4 (#708 scoring, #709 summary trees, #710 retrieval) build on
+//! top of these chunks without modifying the Phase 1 surface.
+
+pub mod canonicalize;
+pub mod chunker;
+pub mod ingest;
+pub mod rpc;
+pub mod schemas;
+pub mod store;
+pub mod types;
+
+pub use schemas::{
+    all_controller_schemas as all_memory_tree_controller_schemas,
+    all_registered_controllers as all_memory_tree_registered_controllers,
+};

src/openhuman/memory/tree/rpc.rs

@@ -0,0 +1,181 @@
+//! RPC handler functions for the memory tree layer.
+//!
+//! Public JSON-RPC surface:
+//! - `openhuman.memory_tree_ingest` — one unified ingest. Caller supplies
+//!   `source_kind` + generic JSON `payload` (adapter-specific). Internally
+//!   dispatches to chat / email / document canonicalisers.
+//! - `openhuman.memory_tree_list_chunks` — listing with filters.
+//! - `openhuman.memory_tree_get_chunk` — single chunk fetch.
+
+use serde::{Deserialize, Serialize};
+use serde_json::Value;
+
+use crate::openhuman::config::Config;
+use crate::openhuman::memory::tree::canonicalize::{
+    chat::ChatBatch, document::DocumentInput, email::EmailThread,
+};
+use crate::openhuman::memory::tree::ingest::{
+    ingest_chat as do_ingest_chat, ingest_document as do_ingest_document,
+    ingest_email as do_ingest_email, IngestResult,
+};
+use crate::openhuman::memory::tree::store::{self, ListChunksQuery};
+use crate::openhuman::memory::tree::types::{Chunk, SourceKind};
+use crate::rpc::RpcOutcome;
+
+/// Unified ingest request. The `payload` shape is adapter-specific and is
+/// validated inside the dispatch based on `source_kind`.
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub struct IngestRequest {
+    /// Which kind of source the payload represents.
+    pub source_kind: SourceKind,
+    /// Logical source id (channel/group for chat, thread for email, doc id).
+    pub source_id: String,
+    /// Account/user this content belongs to.
+    #[serde(default)]
+    pub owner: String,
+    /// Optional labels/tags carried through.
+    #[serde(default)]
+    pub tags: Vec<String>,
+    /// Adapter-specific payload — shape matches the canonicaliser for
+    /// `source_kind`:
+    /// - `chat`     → [`ChatBatch`]
+    /// - `email`    → [`EmailThread`]
+    /// - `document` → [`DocumentInput`]
+    pub payload: Value,
+}
+
+/// Unified ingest RPC handler. Dispatches on `source_kind`.
+pub async fn ingest_rpc(
+    config: &Config,
+    req: IngestRequest,
+) -> Result<RpcOutcome<IngestResult>, String> {
+    let IngestRequest {
+        source_kind,
+        source_id,
+        owner,
+        tags,
+        payload,
+    } = req;
+
+    log::debug!(
+        "[memory_tree::rpc] ingest kind={} source_id={}",
+        source_kind.as_str(),
+        source_id
+    );
+
+    let result = tokio::task::spawn_blocking({
+        let config = config.clone();
+        let source_id = source_id.clone();
+        let owner = owner.clone();
+        move || -> anyhow::Result<IngestResult> {
+            match source_kind {
+                SourceKind::Chat => {
+                    let batch: ChatBatch = serde_json::from_value(payload)
+                        .map_err(|e| anyhow::anyhow!("invalid chat payload: {e}"))?;
+                    do_ingest_chat(&config, &source_id, &owner, tags, batch)
+                }
+                SourceKind::Email => {
+                    let thread: EmailThread = serde_json::from_value(payload)
+                        .map_err(|e| anyhow::anyhow!("invalid email payload: {e}"))?;
+                    do_ingest_email(&config, &source_id, &owner, tags, thread)
+                }
+                SourceKind::Document => {
+                    let doc: DocumentInput = serde_json::from_value(payload)
+                        .map_err(|e| anyhow::anyhow!("invalid document payload: {e}"))?;
+                    do_ingest_document(&config, &source_id, &owner, tags, doc)
+                }
+            }
+        }
+    })
+    .await
+    .map_err(|e| format!("ingest join error: {e}"))?
+    .map_err(|e| format!("ingest: {e}"))?;
+
+    Ok(RpcOutcome::single_log(
+        result,
+        format!(
+            "memory_tree: ingest kind={} source_id={source_id}",
+            source_kind.as_str()
+        ),
+    ))
+}
+
+/// Query shape for the `list_chunks` RPC.
+#[derive(Clone, Debug, Default, Serialize, Deserialize)]
+pub struct ListChunksRequest {
+    #[serde(default)]
+    pub source_kind: Option<String>,
+    #[serde(default)]
+    pub source_id: Option<String>,
+    #[serde(default)]
+    pub owner: Option<String>,
+    #[serde(default)]
+    pub since_ms: Option<i64>,
+    #[serde(default)]
+    pub until_ms: Option<i64>,
+    #[serde(default)]
+    pub limit: Option<usize>,
+}
+
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub struct ListChunksResponse {
+    pub chunks: Vec<Chunk>,
+}
+
+pub async fn list_chunks_rpc(
+    config: &Config,
+    req: ListChunksRequest,
+) -> Result<RpcOutcome<ListChunksResponse>, String> {
+    let query = ListChunksQuery {
+        source_kind: match req.source_kind.as_deref() {
+            None => None,
+            Some(s) => Some(SourceKind::parse(s)?),
+        },
+        source_id: req.source_id,
+        owner: req.owner,
+        since_ms: req.since_ms,
+        until_ms: req.until_ms,
+        limit: req.limit,
+    };
+    let rows = tokio::task::spawn_blocking({
+        let config = config.clone();
+        move || store::list_chunks(&config, &query)
+    })
+    .await
+    .map_err(|e| format!("list_chunks join error: {e}"))?
+    .map_err(|e| format!("list_chunks: {e}"))?;
+
+    let n = rows.len();
+    Ok(RpcOutcome::single_log(
+        ListChunksResponse { chunks: rows },
+        format!("memory_tree: list_chunks n={n}"),
+    ))
+}
+
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub struct GetChunkRequest {
+    pub id: String,
+}
+
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub struct GetChunkResponse {
+    pub chunk: Option<Chunk>,
+}
+
+pub async fn get_chunk_rpc(
+    config: &Config,
+    req: GetChunkRequest,
+) -> Result<RpcOutcome<GetChunkResponse>, String> {
+    let id = req.id.clone();
+    let chunk = tokio::task::spawn_blocking({
+        let config = config.clone();
+        move || store::get_chunk(&config, &id)
+    })
+    .await
+    .map_err(|e| format!("get_chunk join error: {e}"))?
+    .map_err(|e| format!("get_chunk: {e}"))?;
+    Ok(RpcOutcome::single_log(
+        GetChunkResponse { chunk },
+        format!("memory_tree: get_chunk id={}", req.id),
+    ))
+}

src/openhuman/memory/tree/schemas.rs

@@ -0,0 +1,231 @@
+//! Controller schemas for the memory tree (Phase 1 / #707).
+//!
+//! Registered JSON-RPC methods:
+//! - `openhuman.memory_tree_ingest`      — unified ingest (source_kind + JSON payload)
+//! - `openhuman.memory_tree_list_chunks`
+//! - `openhuman.memory_tree_get_chunk`
+//!
+//! Handlers delegate to [`super::rpc`].
+
+use serde::de::DeserializeOwned;
+use serde_json::{Map, Value};
+
+use crate::core::all::{ControllerFuture, RegisteredController};
+use crate::core::{ControllerSchema, FieldSchema, TypeSchema};
+use crate::openhuman::config::rpc as config_rpc;
+use crate::openhuman::memory::tree::rpc as tree_rpc;
+use crate::rpc::RpcOutcome;
+
+const NAMESPACE: &str = "memory_tree";
+
+pub fn all_controller_schemas() -> Vec<ControllerSchema> {
+    vec![
+        schemas("ingest"),
+        schemas("list_chunks"),
+        schemas("get_chunk"),
+    ]
+}
+
+pub fn all_registered_controllers() -> Vec<RegisteredController> {
+    vec![
+        RegisteredController {
+            schema: schemas("ingest"),
+            handler: handle_ingest,
+        },
+        RegisteredController {
+            schema: schemas("list_chunks"),
+            handler: handle_list_chunks,
+        },
+        RegisteredController {
+            schema: schemas("get_chunk"),
+            handler: handle_get_chunk,
+        },
+    ]
+}
+
+pub fn schemas(function: &str) -> ControllerSchema {
+    match function {
+        "ingest" => ControllerSchema {
+            namespace: NAMESPACE,
+            function: "ingest",
+            description: "Ingest a source into canonical chunks. \
+                 Dispatches on `source_kind`; `payload` shape depends on the kind \
+                 (chat → ChatBatch, email → EmailThread, document → DocumentInput).",
+            inputs: vec![
+                FieldSchema {
+                    name: "source_kind",
+                    ty: TypeSchema::Enum {
+                        variants: vec!["chat", "email", "document"],
+                    },
+                    comment: "Which source kind the payload represents.",
+                    required: true,
+                },
+                FieldSchema {
+                    name: "source_id",
+                    ty: TypeSchema::String,
+                    comment: "Stable logical source id (channel, thread, document id).",
+                    required: true,
+                },
+                FieldSchema {
+                    name: "owner",
+                    ty: TypeSchema::String,
+                    comment: "Optional account / user this content belongs to.",
+                    required: false,
+                },
+                FieldSchema {
+                    name: "tags",
+                    ty: TypeSchema::Array(Box::new(TypeSchema::String)),
+                    comment: "Optional tags or labels carried through.",
+                    required: false,
+                },
+                FieldSchema {
+                    name: "payload",
+                    ty: TypeSchema::Json,
+                    comment: "Adapter-specific payload. \
+                         chat: {platform, channel_label, messages[]}. \
+                         email: {provider, thread_subject, messages[]}. \
+                         document: {provider, title, body, modified_at, source_ref}.",
+                    required: true,
+                },
+            ],
+            outputs: vec![
+                FieldSchema {
+                    name: "source_id",
+                    ty: TypeSchema::String,
+                    comment: "Logical source id the ingest was scoped to.",
+                    required: true,
+                },
+                FieldSchema {
+                    name: "chunks_written",
+                    ty: TypeSchema::U64,
+                    comment: "Number of chunks persisted (including idempotent rewrites).",
+                    required: true,
+                },
+                FieldSchema {
+                    name: "chunk_ids",
+                    ty: TypeSchema::Array(Box::new(TypeSchema::String)),
+                    comment: "IDs of all chunks produced.",
+                    required: true,
+                },
+            ],
+        },
+        "list_chunks" => ControllerSchema {
+            namespace: NAMESPACE,
+            function: "list_chunks",
+            description:
+                "List stored chunks, newest first, optionally filtered by source / owner / time.",
+            inputs: vec![
+                FieldSchema {
+                    name: "source_kind",
+                    ty: TypeSchema::Option(Box::new(TypeSchema::Enum {
+                        variants: vec!["chat", "email", "document"],
+                    })),
+                    comment: "Restrict to a single source kind.",
+                    required: false,
+                },
+                FieldSchema {
+                    name: "source_id",
+                    ty: TypeSchema::Option(Box::new(TypeSchema::String)),
+                    comment: "Restrict to a single logical source (channel/thread/doc id).",
+                    required: false,
+                },
+                FieldSchema {
+                    name: "owner",
+                    ty: TypeSchema::Option(Box::new(TypeSchema::String)),
+                    comment: "Restrict to a single owner/account.",
+                    required: false,
+                },
+                FieldSchema {
+                    name: "since_ms",
+                    ty: TypeSchema::Option(Box::new(TypeSchema::I64)),
+                    comment: "Inclusive lower bound on chunk timestamp (ms since epoch).",
+                    required: false,
+                },
+                FieldSchema {
+                    name: "until_ms",
+                    ty: TypeSchema::Option(Box::new(TypeSchema::I64)),
+                    comment: "Inclusive upper bound on chunk timestamp (ms since epoch).",
+                    required: false,
+                },
+                FieldSchema {
+                    name: "limit",
+                    ty: TypeSchema::Option(Box::new(TypeSchema::U64)),
+                    comment: "Maximum rows to return (defaults to 100).",
+                    required: false,
+                },
+            ],
+            outputs: vec![FieldSchema {
+                name: "chunks",
+                ty: TypeSchema::Array(Box::new(TypeSchema::Ref("Chunk"))),
+                comment: "Matching chunks ordered by timestamp DESC.",
+                required: true,
+            }],
+        },
+        "get_chunk" => ControllerSchema {
+            namespace: NAMESPACE,
+            function: "get_chunk",
+            description: "Fetch a single chunk by its deterministic id.",
+            inputs: vec![FieldSchema {
+                name: "id",
+                ty: TypeSchema::String,
+                comment: "Chunk id (32 hex chars).",
+                required: true,
+            }],
+            outputs: vec![FieldSchema {
+                name: "chunk",
+                ty: TypeSchema::Option(Box::new(TypeSchema::Ref("Chunk"))),
+                comment: "The chunk if found, otherwise null.",
+                required: false,
+            }],
+        },
+        _ => ControllerSchema {
+            namespace: NAMESPACE,
+            function: "unknown",
+            description: "Unknown memory_tree controller function.",
+            inputs: vec![FieldSchema {
+                name: "function",
+                ty: TypeSchema::String,
+                comment: "Unknown function requested for schema lookup.",
+                required: true,
+            }],
+            outputs: vec![FieldSchema {
+                name: "error",
+                ty: TypeSchema::String,
+                comment: "Lookup error details.",
+                required: true,
+            }],
+        },
+    }
+}
+
+fn handle_ingest(params: Map<String, Value>) -> ControllerFuture {
+    Box::pin(async move {
+        let config = config_rpc::load_config_with_timeout().await?;
+        let req = parse_value::<tree_rpc::IngestRequest>(Value::Object(params))?;
+        to_json(tree_rpc::ingest_rpc(&config, req).await?)
+    })
+}
+
+fn handle_list_chunks(params: Map<String, Value>) -> ControllerFuture {
+    Box::pin(async move {
+        let config = config_rpc::load_config_with_timeout().await?;
+        let req = parse_value::<tree_rpc::ListChunksRequest>(Value::Object(params))?;
+        to_json(tree_rpc::list_chunks_rpc(&config, req).await?)
+    })
+}
+
+fn handle_get_chunk(params: Map<String, Value>) -> ControllerFuture {
+    Box::pin(async move {
+        let config = config_rpc::load_config_with_timeout().await?;
+        let req = parse_value::<tree_rpc::GetChunkRequest>(Value::Object(params))?;
+        to_json(tree_rpc::get_chunk_rpc(&config, req).await?)
+    })
+}
+
+fn parse_value<T: DeserializeOwned>(v: Value) -> Result<T, String> {
+    serde_json::from_value(v).map_err(|e| format!("invalid params: {e}"))
+}
+
+fn to_json<T: serde::Serialize>(outcome: RpcOutcome<T>) -> Result<Value, String> {
+    outcome.into_cli_compatible_json()
+}

src/openhuman/memory/tree/store.rs

@@ -0,0 +1,414 @@
+//! SQLite-backed persistence for ingested chunks (Phase 1 / issue #707).
+//!
+//! The store lives at `<workspace>/memory_tree/chunks.db`. Schema is applied
+//! lazily on first access via `with_connection`, so the DB is created on
+//! demand without an explicit migration step.
+//!
+//! Upsert semantics: writes are idempotent on `chunk.id` so re-ingesting the
+//! same raw source yields no duplicates.
+
+use anyhow::{Context, Result};
+use chrono::{DateTime, TimeZone, Utc};
+use rusqlite::{params, Connection, OptionalExtension};
+use std::time::Duration;
+
+use crate::openhuman::config::Config;
+use crate::openhuman::memory::tree::types::{Chunk, Metadata, SourceKind, SourceRef};
+
+const DB_DIR: &str = "memory_tree";
+const DB_FILE: &str = "chunks.db";
+const DEFAULT_LIST_LIMIT: usize = 100;
+const MAX_LIST_LIMIT: usize = 10_000;
+const SQLITE_BUSY_TIMEOUT: Duration = Duration::from_secs(5);
+
+const SCHEMA: &str = "
+PRAGMA foreign_keys = ON;
+
+CREATE TABLE IF NOT EXISTS mem_tree_chunks (
+    id                     TEXT PRIMARY KEY,
+    source_kind            TEXT NOT NULL,
+    source_id              TEXT NOT NULL,
+    source_ref             TEXT,
+    owner                  TEXT NOT NULL,
+    timestamp_ms           INTEGER NOT NULL,
+    time_range_start_ms    INTEGER NOT NULL,
+    time_range_end_ms      INTEGER NOT NULL,
+    tags_json              TEXT NOT NULL DEFAULT '[]',
+    content                TEXT NOT NULL,
+    token_count            INTEGER NOT NULL,
+    seq_in_source          INTEGER NOT NULL,
+    created_at_ms          INTEGER NOT NULL
+);
+
+CREATE INDEX IF NOT EXISTS idx_mem_tree_chunks_source
+    ON mem_tree_chunks(source_kind, source_id);
+CREATE INDEX IF NOT EXISTS idx_mem_tree_chunks_timestamp
+    ON mem_tree_chunks(timestamp_ms);
+CREATE INDEX IF NOT EXISTS idx_mem_tree_chunks_owner
+    ON mem_tree_chunks(owner);
+CREATE INDEX IF NOT EXISTS idx_mem_tree_chunks_source_seq
+    ON mem_tree_chunks(source_kind, source_id, seq_in_source);
+";
+
+/// Upsert a batch of chunks atomically.
+///
+/// Returns the number of rows inserted or replaced. Duplicates on `chunk.id`
+/// are replaced, making the operation idempotent for re-ingest of the same
+/// raw source.
+pub fn upsert_chunks(config: &Config, chunks: &[Chunk]) -> Result<usize> {
+    if chunks.is_empty() {
+        return Ok(0);
+    }
+    log::debug!(
+        "[memory_tree::store] upsert_chunks: n={} first_id={}",
+        chunks.len(),
+        chunks[0].id
+    );
+    with_connection(config, |conn| {
+        let tx = conn.unchecked_transaction()?;
+        {
+            let mut stmt = tx.prepare(
+                "INSERT OR REPLACE INTO mem_tree_chunks (
+                    id, source_kind, source_id, source_ref, owner,
+                    timestamp_ms, time_range_start_ms, time_range_end_ms,
+                    tags_json, content, token_count, seq_in_source, created_at_ms
+                ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13)",
+            )?;
+            for chunk in chunks {
+                stmt.execute(params![
+                    chunk.id,
+                    chunk.metadata.source_kind.as_str(),
+                    chunk.metadata.source_id,
+                    chunk.metadata.source_ref.as_ref().map(|r| r.value.as_str()),
+                    chunk.metadata.owner,
+                    chunk.metadata.timestamp.timestamp_millis(),
+                    chunk.metadata.time_range.0.timestamp_millis(),
+                    chunk.metadata.time_range.1.timestamp_millis(),
+                    serde_json::to_string(&chunk.metadata.tags)?,
+                    chunk.content,
+                    chunk.token_count,
+                    chunk.seq_in_source,
+                    chunk.created_at.timestamp_millis(),
+                ])?;
+            }
+        }
+        tx.commit()?;
+        Ok(chunks.len())
+    })
+}
+
+/// Fetch one chunk by its id.
+pub fn get_chunk(config: &Config, id: &str) -> Result<Option<Chunk>> {
+    with_connection(config, |conn| {
+        let mut stmt = conn.prepare(
+            "SELECT id, source_kind, source_id, source_ref, owner,
+                    timestamp_ms, time_range_start_ms, time_range_end_ms,
+                    tags_json, content, token_count, seq_in_source, created_at_ms
+               FROM mem_tree_chunks WHERE id = ?1",
+        )?;
+        let row = stmt
+            .query_row(params![id], row_to_chunk)
+            .optional()
+            .context("Failed to query chunk by id")?;
+        Ok(row)
+    })
+}
+
+/// Query parameters for [`list_chunks`]. All fields are optional filters —
+/// callers pass `ListChunksQuery::default()` to get recent-across-everything.
+#[derive(Debug, Default, Clone)]
+pub struct ListChunksQuery {
+    pub source_kind: Option<SourceKind>,
+    pub source_id: Option<String>,
+    pub owner: Option<String>,
+    /// Inclusive lower bound on `timestamp` (milliseconds since epoch).
+    pub since_ms: Option<i64>,
+    /// Inclusive upper bound on `timestamp` (milliseconds since epoch).
+    pub until_ms: Option<i64>,
+    /// Max rows to return (default 100 when `None`).
+    pub limit: Option<usize>,
+}
+
+/// List chunks matching the provided filters, ordered by `timestamp` DESC.
+pub fn list_chunks(config: &Config, query: &ListChunksQuery) -> Result<Vec<Chunk>> {
+    with_connection(config, |conn| {
+        let mut sql = String::from(
+            "SELECT id, source_kind, source_id, source_ref, owner,
+                    timestamp_ms, time_range_start_ms, time_range_end_ms,
+                    tags_json, content, token_count, seq_in_source, created_at_ms
+               FROM mem_tree_chunks WHERE 1=1",
+        );
+        let mut bound: Vec<Box<dyn rusqlite::ToSql>> = Vec::new();
+
+        if let Some(kind) = query.source_kind {
+            sql.push_str(" AND source_kind = ?");
+            bound.push(Box::new(kind.as_str().to_string()));
+        }
+        if let Some(ref source_id) = query.source_id {
+            sql.push_str(" AND source_id = ?");
+            bound.push(Box::new(source_id.clone()));
+        }
+        if let Some(ref owner) = query.owner {
+            sql.push_str(" AND owner = ?");
+            bound.push(Box::new(owner.clone()));
+        }
+        if let Some(since_ms) = query.since_ms {
+            sql.push_str(" AND timestamp_ms >= ?");
+            bound.push(Box::new(since_ms));
+        }
+        if let Some(until_ms) = query.until_ms {
+            sql.push_str(" AND timestamp_ms <= ?");
+            bound.push(Box::new(until_ms));
+        }
+        let limit = normalized_limit(query.limit);
+        sql.push_str(" ORDER BY timestamp_ms DESC, seq_in_source ASC LIMIT ?");
+        bound.push(Box::new(limit));
+
+        let mut stmt = conn.prepare(&sql)?;
+        let param_refs: Vec<&dyn rusqlite::ToSql> = bound
+            .iter()
+            .map(|b| b.as_ref() as &dyn rusqlite::ToSql)
+            .collect();
+        let rows = stmt
+            .query_map(param_refs.as_slice(), row_to_chunk)?
+            .collect::<rusqlite::Result<Vec<_>>>()
+            .context("Failed to collect chunks")?;
+        Ok(rows)
+    })
+}
+
+/// Count total chunks in the store (useful for tests / diagnostics).
+pub fn count_chunks(config: &Config) -> Result<u64> {
+    with_connection(config, |conn| {
+        let n: i64 = conn.query_row("SELECT COUNT(*) FROM mem_tree_chunks", [], |r| r.get(0))?;
+        Ok(n.max(0) as u64)
+    })
+}
+
+fn row_to_chunk(row: &rusqlite::Row<'_>) -> rusqlite::Result<Chunk> {
+    let id: String = row.get(0)?;
+    let source_kind_s: String = row.get(1)?;
+    let source_id: String = row.get(2)?;
+    let source_ref: Option<String> = row.get(3)?;
+    let owner: String = row.get(4)?;
+    let ts_ms: i64 = row.get(5)?;
+    let trs_ms: i64 = row.get(6)?;
+    let tre_ms: i64 = row.get(7)?;
+    let tags_json: String = row.get(8)?;
+    let content: String = row.get(9)?;
+    let token_count: i64 = row.get(10)?;
+    let seq: i64 = row.get(11)?;
+    let created_ms: i64 = row.get(12)?;
+
+    let source_kind = SourceKind::parse(&source_kind_s).map_err(|e| {
+        rusqlite::Error::FromSqlConversionFailure(1, rusqlite::types::Type::Text, e.into())
+    })?;
+    let timestamp = ms_to_utc(ts_ms)?;
+    let time_range = (ms_to_utc(trs_ms)?, ms_to_utc(tre_ms)?);
+    let created_at = ms_to_utc(created_ms)?;
+    let tags: Vec<String> = serde_json::from_str(&tags_json).map_err(|e| {
+        rusqlite::Error::FromSqlConversionFailure(8, rusqlite::types::Type::Text, Box::new(e))
+    })?;
+
+    Ok(Chunk {
+        id,
+        content,
+        metadata: Metadata {
+            source_kind,
+            source_id,
+            owner,
+            timestamp,
+            time_range,
+            tags,
+            source_ref: source_ref.map(SourceRef::new),
+        },
+        token_count: token_count.max(0) as u32,
+        seq_in_source: seq.max(0) as u32,
+        created_at,
+    })
+}
+
+fn ms_to_utc(ms: i64) -> rusqlite::Result<DateTime<Utc>> {
+    Utc.timestamp_millis_opt(ms).single().ok_or_else(|| {
+        rusqlite::Error::FromSqlConversionFailure(
+            0,
+            rusqlite::types::Type::Integer,
+            format!("invalid timestamp ms {ms}").into(),
+        )
+    })
+}
+
+fn with_connection<T>(config: &Config, f: impl FnOnce(&Connection) -> Result<T>) -> Result<T> {
+    let dir = config.workspace_dir.join(DB_DIR);
+    std::fs::create_dir_all(&dir)
+        .with_context(|| format!("Failed to create memory_tree dir: {}", dir.display()))?;
+    let db_path = dir.join(DB_FILE);
+    let conn = Connection::open(&db_path)
+        .with_context(|| format!("Failed to open memory_tree DB: {}", db_path.display()))?;
+    conn.busy_timeout(SQLITE_BUSY_TIMEOUT)
+        .context("Failed to configure memory_tree busy timeout")?;
+    conn.execute_batch("PRAGMA journal_mode=WAL;")
+        .context("Failed to enable memory_tree WAL mode")?;
+    conn.execute_batch(SCHEMA)
+        .context("Failed to initialize memory_tree schema")?;
+    f(&conn)
+}
+
+fn normalized_limit(requested: Option<usize>) -> i64 {
+    let clamped = requested
+        .unwrap_or(DEFAULT_LIST_LIMIT)
+        .clamp(1, MAX_LIST_LIMIT);
+    i64::try_from(clamped).unwrap_or(MAX_LIST_LIMIT as i64)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::openhuman::memory::tree::types::chunk_id;
+    use chrono::TimeZone;
+    use tempfile::TempDir;
+
+    fn test_config() -> (TempDir, Config) {
+        let tmp = TempDir::new().expect("tempdir");
+        let mut cfg = Config::default();
+        cfg.workspace_dir = tmp.path().to_path_buf();
+        (tmp, cfg)
+    }
+
+    fn sample_chunk(source_id: &str, seq: u32, ts_ms: i64) -> Chunk {
+        let ts = Utc.timestamp_millis_opt(ts_ms).unwrap();
+        Chunk {
+            id: chunk_id(SourceKind::Chat, source_id, seq),
+            content: format!("content {source_id} {seq}"),
+            metadata: Metadata {
+                source_kind: SourceKind::Chat,
+                source_id: source_id.to_string(),
+                owner: "alice@example.com".to_string(),
+                timestamp: ts,
+                time_range: (ts, ts),
+                tags: vec!["eng".into()],
+                source_ref: Some(SourceRef::new(format!("slack://{source_id}/{seq}"))),
+            },
+            token_count: 12,
+            seq_in_source: seq,
+            created_at: ts,
+        }
+    }
+
+    #[test]
+    fn upsert_then_get() {
+        let (_tmp, cfg) = test_config();
+        let c = sample_chunk("slack:#eng", 0, 1_700_000_000_000);
+        assert_eq!(upsert_chunks(&cfg, &[c.clone()]).unwrap(), 1);
+        let got = get_chunk(&cfg, &c.id).unwrap().expect("chunk stored");
+        assert_eq!(got, c);
+    }
+
+    #[test]
+    fn upsert_is_idempotent() {
+        let (_tmp, cfg) = test_config();
+        let c = sample_chunk("slack:#eng", 0, 1_700_000_000_000);
+        upsert_chunks(&cfg, &[c.clone()]).unwrap();
+        upsert_chunks(&cfg, &[c.clone()]).unwrap();
+        assert_eq!(count_chunks(&cfg).unwrap(), 1);
+    }
+
+    #[test]
+    fn list_filters_by_source_kind() {
+        let (_tmp, cfg) = test_config();
+        let c1 = sample_chunk("slack:#eng", 0, 1_700_000_000_000);
+        let mut c2 = sample_chunk("gmail:t1", 0, 1_700_000_001_000);
+        c2.metadata.source_kind = SourceKind::Email;
+        upsert_chunks(&cfg, &[c1.clone(), c2.clone()]).unwrap();
+        let q = ListChunksQuery {
+            source_kind: Some(SourceKind::Email),
+            ..Default::default()
+        };
+        let rows = list_chunks(&cfg, &q).unwrap();
+        assert_eq!(rows.len(), 1);
+        assert_eq!(rows[0].metadata.source_kind, SourceKind::Email);
+    }
+
+    #[test]
+    fn list_filters_by_time_range() {
+        let (_tmp, cfg) = test_config();
+        let a = sample_chunk("s", 0, 1_700_000_000_000);
+        let b = sample_chunk("s", 1, 1_700_000_010_000);
+        let c = sample_chunk("s", 2, 1_700_000_020_000);
+        upsert_chunks(&cfg, &[a.clone(), b.clone(), c.clone()]).unwrap();
+        let q = ListChunksQuery {
+            since_ms: Some(1_700_000_005_000),
+            until_ms: Some(1_700_000_015_000),
+            ..Default::default()
+        };
+        let rows = list_chunks(&cfg, &q).unwrap();
+        assert_eq!(rows.len(), 1);
+        assert_eq!(rows[0].id, b.id);
+    }
+
+    #[test]
+    fn list_orders_by_timestamp_desc() {
+        let (_tmp, cfg) = test_config();
+        let a = sample_chunk("s", 0, 1_700_000_000_000);
+        let b = sample_chunk("s", 1, 1_700_000_010_000);
+        upsert_chunks(&cfg, &[a.clone(), b.clone()]).unwrap();
+        let rows = list_chunks(&cfg, &ListChunksQuery::default()).unwrap();
+        assert_eq!(rows.len(), 2);
+        assert_eq!(rows[0].id, b.id); // newest first
+        assert_eq!(rows[1].id, a.id);
+    }
+
+    #[test]
+    fn list_orders_equal_timestamps_by_sequence() {
+        let (_tmp, cfg) = test_config();
+        let a = sample_chunk("s", 0, 1_700_000_000_000);
+        let b = sample_chunk("s", 1, 1_700_000_000_000);
+        upsert_chunks(&cfg, &[b.clone(), a.clone()]).unwrap();
+        let rows = list_chunks(&cfg, &ListChunksQuery::default()).unwrap();
+        assert_eq!(rows.len(), 2);
+        assert_eq!(rows[0].seq_in_source, 0);
+        assert_eq!(rows[1].seq_in_source, 1);
+    }
+
+    #[test]
+    fn list_limit_is_clamped_to_sane_range() {
+        let (_tmp, cfg) = test_config();
+        let chunks = (0..3)
+            .map(|idx| sample_chunk("s", idx, 1_700_000_000_000 + i64::from(idx)))
+            .collect::<Vec<_>>();
+        upsert_chunks(&cfg, &chunks).unwrap();
+
+        let zero_limit = list_chunks(
+            &cfg,
+            &ListChunksQuery {
+                limit: Some(0),
+                ..Default::default()
+            },
+        )
+        .unwrap();
+        assert_eq!(zero_limit.len(), 1);
+
+        let huge_limit = list_chunks(
+            &cfg,
+            &ListChunksQuery {
+                limit: Some(usize::MAX),
+                ..Default::default()
+            },
+        )
+        .unwrap();
+        assert_eq!(huge_limit.len(), 3);
+    }
+
+    #[test]
+    fn missing_chunk_returns_none() {
+        let (_tmp, cfg) = test_config();
+        assert!(get_chunk(&cfg, "nonexistent").unwrap().is_none());
+    }
+
+    #[test]
+    fn empty_batch_is_noop() {
+        let (_tmp, cfg) = test_config();
+        assert_eq!(upsert_chunks(&cfg, &[]).unwrap(), 0);
+        assert_eq!(count_chunks(&cfg).unwrap(), 0);
+    }
+}

src/openhuman/memory/tree/types.rs

@@ -0,0 +1,397 @@
+//! Core types for the memory tree ingestion layer (Phase 1 / issue #707).
+//!
+//! This module defines the canonical [`Chunk`] representation produced by the
+//! ingestion pipeline along with its provenance [`Metadata`] and back-pointer
+//! [`SourceRef`]. These types feed into later phases (#708 scoring, #709
+//! summary trees, #710 retrieval) but are self-contained at Phase 1.
+//!
+//! All chunk IDs are deterministic: `sha256(source_kind | "\0" | source_id |
+//! "\0" | seq)` truncated to 32 hex chars so re-ingest of the same source
+//! material yields stable IDs and idempotent upserts.
+
+use chrono::{DateTime, Utc};
+use serde::{Deserialize, Serialize};
+use sha2::{Digest, Sha256};
+
+/// Which kind of upstream source produced a chunk.
+///
+/// Used both as a metadata discriminator and as the routing key for the
+/// canonicaliser dispatch in [`super::canonicalize`].
+#[derive(Clone, Copy, Debug, Eq, PartialEq, Hash, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum SourceKind {
+    /// Chat transcript scoped by channel or group (Slack, Discord, Telegram, WhatsApp…).
+    Chat,
+    /// Email thread (Gmail and generic IMAP).
+    Email,
+    /// Standalone document (Notion page, Drive doc, meeting note, uploaded file…).
+    Document,
+}
+
+impl SourceKind {
+    /// Stable string representation for DB storage and RPC surfaces.
+    pub fn as_str(self) -> &'static str {
+        match self {
+            SourceKind::Chat => "chat",
+            SourceKind::Email => "email",
+            SourceKind::Document => "document",
+        }
+    }
+
+    /// Parse back from the on-wire / on-disk string form.
+    pub fn parse(s: &str) -> Result<Self, String> {
+        match s {
+            "chat" => Ok(SourceKind::Chat),
+            "email" => Ok(SourceKind::Email),
+            "document" => Ok(SourceKind::Document),
+            other => Err(format!("unknown source kind: {other}")),
+        }
+    }
+}
+
+/// Concrete upstream provider the content came from.
+///
+/// Enumerates every provider listed in `m.excalidraw` Step 1 — Collect the
+/// Data. Each variant maps to exactly one [`SourceKind`] via [`Self::kind`].
+///
+/// Wire form is snake_case (see `as_str` / `parse`) so it is stable across
+/// DB rows, JSON-RPC payloads, and logs.
+///
+/// Marked `#[non_exhaustive]` so new providers can be added in later phases
+/// without breaking downstream pattern matches.
+#[derive(Clone, Copy, Debug, Eq, PartialEq, Hash, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+#[non_exhaustive]
+pub enum DataSource {
+    // ── Chat transcripts (grouped by channel/group) ────────────────────
+    Discord,
+    Telegram,
+    Whatsapp,
+
+    // ── Email threads (grouped by thread) ──────────────────────────────
+    Gmail,
+    /// Catch-all for non-Gmail providers (Outlook, FastMail, generic IMAP, …).
+    OtherEmail,
+
+    // ── Documents (no grouping) ────────────────────────────────────────
+    Notion,
+    MeetingNotes,
+    DriveDocs,
+}
+
+impl DataSource {
+    /// Which [`SourceKind`] this provider feeds into.
+    pub fn kind(self) -> SourceKind {
+        match self {
+            Self::Discord | Self::Telegram | Self::Whatsapp => SourceKind::Chat,
+            Self::Gmail | Self::OtherEmail => SourceKind::Email,
+            Self::Notion | Self::MeetingNotes | Self::DriveDocs => SourceKind::Document,
+        }
+    }
+
+    /// Stable snake_case identifier for DB storage, RPC payloads, and logs.
+    pub fn as_str(self) -> &'static str {
+        match self {
+            Self::Discord => "discord",
+            Self::Telegram => "telegram",
+            Self::Whatsapp => "whatsapp",
+            Self::Gmail => "gmail",
+            Self::OtherEmail => "other_email",
+            Self::Notion => "notion",
+            Self::MeetingNotes => "meeting_notes",
+            Self::DriveDocs => "drive_docs",
+        }
+    }
+
+    /// Parse back from the on-wire / on-disk string form.
+    pub fn parse(s: &str) -> Result<Self, String> {
+        match s {
+            "discord" => Ok(Self::Discord),
+            "telegram" => Ok(Self::Telegram),
+            "whatsapp" => Ok(Self::Whatsapp),
+            "gmail" => Ok(Self::Gmail),
+            "other_email" => Ok(Self::OtherEmail),
+            "notion" => Ok(Self::Notion),
+            "meeting_notes" => Ok(Self::MeetingNotes),
+            "drive_docs" => Ok(Self::DriveDocs),
+            other => Err(format!("unknown data source: {other}")),
+        }
+    }
+
+    /// Every known variant, in declaration order.
+    ///
+    /// Useful for tests, CLI completion, and enumerating supported providers
+    /// in diagnostic output.
+    pub fn all() -> &'static [DataSource] {
+        &[
+            Self::Discord,
+            Self::Telegram,
+            Self::Whatsapp,
+            Self::Gmail,
+            Self::OtherEmail,
+            Self::Notion,
+            Self::MeetingNotes,
+            Self::DriveDocs,
+        ]
+    }
+}
+
+/// A concrete pointer back to where a chunk originated — used for citation,
+/// drill-down, and deduplication at re-ingest time.
+///
+/// Consumers should treat this as an opaque, source-specific reference. The
+/// shape depends on [`SourceKind`]:
+/// - **Chat**: `{platform}://{channel}/{message_id}` or `{permalink}`
+/// - **Email**: message-id header (`<abc@example.com>`) or provider URL
+/// - **Document**: file path, Notion page URL, Drive file id
+#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
+pub struct SourceRef {
+    /// Opaque provider-specific identifier for the exact source record.
+    pub value: String,
+}
+
+impl SourceRef {
+    pub fn new(value: impl Into<String>) -> Self {
+        Self {
+            value: value.into(),
+        }
+    }
+}
+
+/// Provenance metadata captured per chunk at ingest time.
+///
+/// Acceptance criteria on #707 require at minimum: source type, source
+/// identifier, owner/account, timestamps, and tags/labels when available.
+#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
+pub struct Metadata {
+    /// Which upstream source kind produced this chunk.
+    pub source_kind: SourceKind,
+    /// Stable logical id for the ingestion group (channel id, thread id, doc id).
+    ///
+    /// Chat: channel/group id. Email: thread id. Document: doc id.
+    pub source_id: String,
+    /// Account or user the content belongs to. Empty string for anonymous / system sources.
+    pub owner: String,
+    /// Point-in-time timestamp for ordering within a source.
+    ///
+    /// For chats = message time; for emails = message sent time;
+    /// for documents = last-modified or ingest time.
+    #[serde(with = "chrono::serde::ts_milliseconds")]
+    pub timestamp: DateTime<Utc>,
+    /// Covering time range the chunk spans. For a single leaf it usually equals
+    /// `(timestamp, timestamp)`; for later summary nodes (#709) it widens to
+    /// cover all children.
+    #[serde(with = "time_range_serde")]
+    pub time_range: (DateTime<Utc>, DateTime<Utc>),
+    /// Arbitrary labels / tags carried through from the source (e.g. Gmail labels,
+    /// Slack reactions, Notion tags). Ingest does not interpret these.
+    #[serde(default)]
+    pub tags: Vec<String>,
+    /// Opaque pointer back to the raw source record for drill-down / citation.
+    pub source_ref: Option<SourceRef>,
+}
+
+impl Metadata {
+    /// Convenience constructor used by canonicalisers: point timestamp,
+    /// `time_range = (timestamp, timestamp)`.
+    pub fn point_in_time(
+        source_kind: SourceKind,
+        source_id: impl Into<String>,
+        owner: impl Into<String>,
+        timestamp: DateTime<Utc>,
+    ) -> Self {
+        Self {
+            source_kind,
+            source_id: source_id.into(),
+            owner: owner.into(),
+            timestamp,
+            time_range: (timestamp, timestamp),
+            tags: Vec::new(),
+            source_ref: None,
+        }
+    }
+}
+
+/// A single ingested chunk — the atomic persistence unit for Phase 1.
+///
+/// In the LLD this is the leaf of a source tree. Later phases will build
+/// summary nodes on top of these leaves; at Phase 1 they live standalone.
+#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
+pub struct Chunk {
+    /// Deterministic id derived from (source_kind, source_id, seq_in_source).
+    pub id: String,
+    /// Canonical Markdown content.
+    pub content: String,
+    /// Provenance metadata.
+    pub metadata: Metadata,
+    /// Token count (rough heuristic — 1 token ≈ 4 chars — at Phase 1).
+    pub token_count: u32,
+    /// Sequence number of this chunk inside its logical source. Stable and
+    /// starts at 0 for the first chunk of a source.
+    pub seq_in_source: u32,
+    /// When this chunk was persisted to the local store.
+    #[serde(with = "chrono::serde::ts_milliseconds")]
+    pub created_at: DateTime<Utc>,
+}
+
+/// Deterministic chunk id.
+///
+/// `sha256(source_kind | "\0" | source_id | "\0" | seq)` hex-encoded, first
+/// 32 chars (128 bits of collision resistance). Short enough for human
+/// inspection, long enough for global uniqueness in a single-user workspace.
+pub fn chunk_id(source_kind: SourceKind, source_id: &str, seq_in_source: u32) -> String {
+    let mut hasher = Sha256::new();
+    hasher.update(source_kind.as_str().as_bytes());
+    hasher.update([0u8]);
+    hasher.update(source_id.as_bytes());
+    hasher.update([0u8]);
+    hasher.update(seq_in_source.to_be_bytes());
+    let digest = hasher.finalize();
+    let hex = digest.iter().fold(String::with_capacity(64), |mut acc, b| {
+        use std::fmt::Write;
+        let _ = write!(acc, "{b:02x}");
+        acc
+    });
+    hex[..32].to_string()
+}
+
+/// Approximate token count (GPT-family heuristic: 1 token ≈ 4 chars).
+///
+/// Phase 1 does not need a real tokenizer — downstream phases (#709) will
+/// enforce the 10k summariser budget with a precise tokenizer.
+pub fn approx_token_count(text: &str) -> u32 {
+    // saturating_add guards against absurdly long inputs
+    let chars = text.chars().count() as u32;
+    chars.saturating_add(3) / 4
+}
+
+mod time_range_serde {
+    use chrono::{DateTime, TimeZone, Utc};
+    use serde::{Deserialize, Deserializer, Serialize, Serializer};
+
+    #[derive(Serialize, Deserialize)]
+    struct Wire {
+        start_ms: i64,
+        end_ms: i64,
+    }
+
+    pub fn serialize<S: Serializer>(
+        value: &(DateTime<Utc>, DateTime<Utc>),
+        serializer: S,
+    ) -> Result<S::Ok, S::Error> {
+        Wire {
+            start_ms: value.0.timestamp_millis(),
+            end_ms: value.1.timestamp_millis(),
+        }
+        .serialize(serializer)
+    }
+
+    pub fn deserialize<'de, D: Deserializer<'de>>(
+        deserializer: D,
+    ) -> Result<(DateTime<Utc>, DateTime<Utc>), D::Error> {
+        let wire = Wire::deserialize(deserializer)?;
+        let start = Utc
+            .timestamp_millis_opt(wire.start_ms)
+            .single()
+            .ok_or_else(|| serde::de::Error::custom("invalid start_ms"))?;
+        let end = Utc
+            .timestamp_millis_opt(wire.end_ms)
+            .single()
+            .ok_or_else(|| serde::de::Error::custom("invalid end_ms"))?;
+        Ok((start, end))
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn chunk_id_is_deterministic() {
+        let a = chunk_id(SourceKind::Chat, "slack:#eng", 0);
+        let b = chunk_id(SourceKind::Chat, "slack:#eng", 0);
+        assert_eq!(a, b);
+        assert_eq!(a.len(), 32);
+    }
+
+    #[test]
+    fn chunk_id_varies_with_seq() {
+        let a = chunk_id(SourceKind::Chat, "slack:#eng", 0);
+        let b = chunk_id(SourceKind::Chat, "slack:#eng", 1);
+        assert_ne!(a, b);
+    }
+
+    #[test]
+    fn chunk_id_varies_with_source_kind() {
+        let a = chunk_id(SourceKind::Chat, "foo", 0);
+        let b = chunk_id(SourceKind::Email, "foo", 0);
+        assert_ne!(a, b);
+    }
+
+    #[test]
+    fn chunk_id_varies_with_source_id() {
+        let a = chunk_id(SourceKind::Chat, "x", 0);
+        let b = chunk_id(SourceKind::Chat, "y", 0);
+        assert_ne!(a, b);
+    }
+
+    #[test]
+    fn source_kind_round_trip() {
+        for kind in [SourceKind::Chat, SourceKind::Email, SourceKind::Document] {
+            assert_eq!(SourceKind::parse(kind.as_str()).unwrap(), kind);
+        }
+    }
+
+    #[test]
+    fn data_source_round_trip() {
+        for ds in DataSource::all() {
+            assert_eq!(DataSource::parse(ds.as_str()).unwrap(), *ds);
+        }
+    }
+
+    #[test]
+    fn data_source_has_all_eight_variants_from_m_excalidraw() {
+        // Guard against accidental drift from the canonical provider list.
+        assert_eq!(DataSource::all().len(), 8);
+    }
+
+    #[test]
+    fn data_source_kind_mapping() {
+        use DataSource::*;
+        for ds in [Discord, Telegram, Whatsapp] {
+            assert_eq!(ds.kind(), SourceKind::Chat);
+        }
+        for ds in [Gmail, OtherEmail] {
+            assert_eq!(ds.kind(), SourceKind::Email);
+        }
+        for ds in [Notion, MeetingNotes, DriveDocs] {
+            assert_eq!(ds.kind(), SourceKind::Document);
+        }
+    }
+
+    #[test]
+    fn data_source_parse_rejects_unknown() {
+        assert!(DataSource::parse("nope").is_err());
+        // Ensure our snake_case wire form is exactly what callers send.
+        assert!(DataSource::parse("Discord").is_err()); // case-sensitive
+        assert!(DataSource::parse("drive docs").is_err()); // no spaces
+    }
+
+    #[test]
+    fn data_source_serde_is_snake_case() {
+        let ds = DataSource::MeetingNotes;
+        let json = serde_json::to_string(&ds).unwrap();
+        assert_eq!(json, "\"meeting_notes\"");
+        let parsed: DataSource = serde_json::from_str("\"meeting_notes\"").unwrap();
+        assert_eq!(parsed, ds);
+    }
+
+    #[test]
+    fn approx_token_count_scales_linearly() {
+        assert_eq!(approx_token_count(""), 0);
+        assert_eq!(approx_token_count("a"), 1); // 1→1
+        assert_eq!(approx_token_count("abcd"), 1); // 4→1
+        assert_eq!(approx_token_count("abcde"), 2); // 5→2
+        assert_eq!(approx_token_count(&"x".repeat(400)), 100);
+    }
+}

tests/json_rpc_e2e.rs

@@ -16,6 +16,7 @@ use serde_json::{json, Value};
 use tempfile::tempdir;
 
 use openhuman_core::core::jsonrpc::build_core_http_router;
+use openhuman_core::openhuman::memory::all_memory_tree_registered_controllers;
 
 struct EnvVarGuard {
     key: &'static str,
@@ -769,6 +770,154 @@ async fn json_rpc_protocol_auth_and_agent_hello() {
     rpc_join.abort();
 }
 
+#[tokio::test]
+async fn json_rpc_memory_tree_end_to_end() {
+    let _env_lock = json_rpc_e2e_env_lock();
+    let tmp = tempdir().expect("tempdir");
+    let home = tmp.path();
+    let openhuman_home = home.join(".openhuman");
+
+    let _home_guard = EnvVarGuard::set_to_path("HOME", home);
+    let _workspace_guard = EnvVarGuard::unset("OPENHUMAN_WORKSPACE");
+    let _backend_url_guard = EnvVarGuard::unset("BACKEND_URL");
+    let _vite_backend_guard = EnvVarGuard::unset("VITE_BACKEND_URL");
+
+    let (mock_addr, mock_join) = serve_on_ephemeral(mock_upstream_router()).await;
+    let mock_origin = format!("http://{}", mock_addr);
+    write_min_config(&openhuman_home, &mock_origin);
+
+    let controllers = all_memory_tree_registered_controllers();
+    let expected_methods = vec![
+        "openhuman.memory_tree_ingest".to_string(),
+        "openhuman.memory_tree_list_chunks".to_string(),
+        "openhuman.memory_tree_get_chunk".to_string(),
+    ];
+    assert_eq!(controllers.len(), expected_methods.len());
+    for method in &expected_methods {
+        assert!(
+            controllers
+                .iter()
+                .any(|controller| controller.rpc_method_name() == *method),
+            "expected memory_tree controller registration for {method}"
+        );
+    }
+
+    let (rpc_addr, rpc_join) = serve_on_ephemeral(build_core_http_router(false)).await;
+    let rpc_base = format!("http://{}", rpc_addr);
+
+    tokio::time::sleep(Duration::from_millis(100)).await;
+
+    let ingest = post_json_rpc(
+        &rpc_base,
+        200,
+        &expected_methods[0],
+        json!({
+            "source_kind": "document",
+            "source_id": "notion:launch-plan",
+            "owner": "alice@example.com",
+            "tags": ["planning", "launch"],
+            "payload": {
+                "provider": "notion",
+                "title": "Launch Plan",
+                "body": "Alpha\n\nBeta",
+                "modified_at": 1700000000000_i64,
+                "source_ref": " notion://page/launch-plan "
+            }
+        }),
+    )
+    .await;
+    let ingest_outer = assert_no_jsonrpc_error(&ingest, "memory_tree_ingest");
+    let ingest_result = ingest_outer.get("result").unwrap_or(ingest_outer);
+    assert_eq!(
+        ingest_result.get("source_id"),
+        Some(&json!("notion:launch-plan"))
+    );
+    assert_eq!(ingest_result.get("chunks_written"), Some(&json!(1)));
+    let chunk_ids = ingest_result
+        .get("chunk_ids")
+        .and_then(Value::as_array)
+        .expect("chunk_ids array");
+    assert_eq!(chunk_ids.len(), 1);
+
+    let list = post_json_rpc(
+        &rpc_base,
+        201,
+        &expected_methods[1],
+        json!({
+            "source_kind": "document",
+            "source_id": "notion:launch-plan",
+            "owner": "alice@example.com",
+            "limit": 0
+        }),
+    )
+    .await;
+    let list_outer = assert_no_jsonrpc_error(&list, "memory_tree_list_chunks");
+    let list_result = list_outer.get("result").unwrap_or(list_outer);
+    let chunks = list_result
+        .get("chunks")
+        .and_then(Value::as_array)
+        .expect("chunks array");
+    assert_eq!(chunks.len(), 1);
+    let chunk = &chunks[0];
+    assert_eq!(chunk.get("seq_in_source"), Some(&json!(0)));
+    assert_eq!(
+        chunk.pointer("/metadata/source_ref/value"),
+        Some(&json!("notion://page/launch-plan"))
+    );
+
+    let get_chunk = post_json_rpc(
+        &rpc_base,
+        202,
+        &expected_methods[2],
+        json!({
+            "id": chunk_ids[0].clone()
+        }),
+    )
+    .await;
+    let get_outer = assert_no_jsonrpc_error(&get_chunk, "memory_tree_get_chunk");
+    let get_result = get_outer.get("result").unwrap_or(get_outer);
+    assert_eq!(get_result.pointer("/chunk/id"), Some(&chunk_ids[0]));
+
+    let invalid_ingest = post_json_rpc(
+        &rpc_base,
+        203,
+        &expected_methods[0],
+        json!({
+            "source_kind": "document",
+            "source_id": "notion:bad",
+            "owner": "alice@example.com",
+            "payload": {
+                "provider": "notion",
+                "title": "Bad payload"
+            }
+        }),
+    )
+    .await;
+    assert!(
+        invalid_ingest.get("error").is_some(),
+        "expected invalid payload JSON-RPC error: {invalid_ingest}"
+    );
+
+    let invalid_list = post_json_rpc(
+        &rpc_base,
+        204,
+        &expected_methods[1],
+        json!({
+            "source_kind": "not-a-kind"
+        }),
+    )
+    .await;
+    assert!(
+        invalid_list.get("error").is_some(),
+        "expected invalid source_kind JSON-RPC error: {invalid_list}"
+    );
+
+    rpc_join.abort();
+    let _ = rpc_join.await;
+    mock_join.abort();
+    let _ = mock_join.await;
+}
+
 #[tokio::test]
 async fn json_rpc_web_chat_routing_cases_use_expected_backend_models() {
     let _env_lock = json_rpc_e2e_env_lock();