diff --git a/src/core/all.rs b/src/core/all.rs index f65942e4c..5965fb795 100644 --- a/src/core/all.rs +++ b/src/core/all.rs @@ -116,6 +116,8 @@ fn build_registered_controllers() -> Vec { 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 { 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()); diff --git a/src/openhuman/memory/mod.rs b/src/openhuman/memory/mod.rs index 40411a7dc..52bab6705 100644 --- a/src/openhuman/memory/mod.rs +++ b/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}; diff --git a/src/openhuman/memory/tree/canonicalize/chat.rs b/src/openhuman/memory/tree/canonicalize/chat.rs new file mode 100644 index 000000000..1a1b91ed9 --- /dev/null +++ b/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, + /// Plain text / markdown body. + pub text: String, + /// Optional per-message provenance pointer (permalink or `platform://...`). + #[serde(default)] + pub source_ref: Option, +} + +/// 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, +} + +/// 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, 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()); + } +} diff --git a/src/openhuman/memory/tree/canonicalize/document.rs b/src/openhuman/memory/tree/canonicalize/document.rs new file mode 100644 index 000000000..1fa6e9e8f --- /dev/null +++ b/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, + /// Optional pointer back to source (URL, file path, Notion page id). + #[serde(default)] + pub source_ref: Option, +} + +pub fn canonicalise( + source_id: &str, + owner: &str, + tags: &[String], + doc: DocumentInput, +) -> Result, 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()); + } +} diff --git a/src/openhuman/memory/tree/canonicalize/email.rs b/src/openhuman/memory/tree/canonicalize/email.rs new file mode 100644 index 000000000..f7d12df54 --- /dev/null +++ b/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, + #[serde(default)] + pub cc: Vec, + pub subject: String, + #[serde(with = "chrono::serde::ts_milliseconds")] + pub sent_at: DateTime, + /// 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, +} + +/// 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, +} + +pub fn canonicalise( + source_id: &str, + owner: &str, + tags: &[String], + thread: EmailThread, +) -> Result, 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!("")), + } + } + + #[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, + "" + ); + } + + #[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()); + } +} diff --git a/src/openhuman/memory/tree/canonicalize/mod.rs b/src/openhuman/memory/tree/canonicalize/mod.rs new file mode 100644 index 000000000..2e57bb29a --- /dev/null +++ b/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

{ + /// 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, +} + +/// Trim provider-specific source references and drop blank pointers. +pub fn normalize_source_ref(source_ref: Option) -> Option { + source_ref.and_then(|value| { + let trimmed = value.trim(); + if trimmed.is_empty() { + None + } else { + Some(SourceRef::new(trimmed.to_string())) + } + }) +} diff --git a/src/openhuman/memory/tree/chunker.rs b/src/openhuman/memory/tree/chunker.rs new file mode 100644 index 000000000..44fedee66 --- /dev/null +++ b/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 { + 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 { + 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(¶graphs, "\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> { + let sep_len = sep.len(); + let mut out: Vec = 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 { + 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::() + ); + } + } + + #[test] + fn falls_back_to_line_split_when_no_paragraphs() { + let text = (0..30) + .map(|i| format!("line-{i}-{}", "x".repeat(40))) + .collect::>() + .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::>(), + b.iter().map(|c| c.id.clone()).collect::>() + ); + } + + #[test] + fn sequence_numbers_start_at_zero() { + let text = (0..10) + .map(|i| format!("p{i}")) + .collect::>() + .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"); + } +} diff --git a/src/openhuman/memory/tree/ingest.rs b/src/openhuman/memory/tree/ingest.rs new file mode 100644 index 000000000..9b68862fd --- /dev/null +++ b/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, +} + +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, + batch: ChatBatch, +) -> Result { + 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, + thread: EmailThread, +) -> Result { + 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, + doc: DocumentInput, +) -> Result { + 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 { + 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" + ); + } +} diff --git a/src/openhuman/memory/tree/mod.rs b/src/openhuman/memory/tree/mod.rs new file mode 100644 index 000000000..7c2ac7bcd --- /dev/null +++ b/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, +}; diff --git a/src/openhuman/memory/tree/rpc.rs b/src/openhuman/memory/tree/rpc.rs new file mode 100644 index 000000000..d420efe04 --- /dev/null +++ b/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, + /// 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, 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 { + 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, + #[serde(default)] + pub source_id: Option, + #[serde(default)] + pub owner: Option, + #[serde(default)] + pub since_ms: Option, + #[serde(default)] + pub until_ms: Option, + #[serde(default)] + pub limit: Option, +} + +#[derive(Clone, Debug, Serialize, Deserialize)] +pub struct ListChunksResponse { + pub chunks: Vec, +} + +pub async fn list_chunks_rpc( + config: &Config, + req: ListChunksRequest, +) -> Result, 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, +} + +pub async fn get_chunk_rpc( + config: &Config, + req: GetChunkRequest, +) -> Result, 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), + )) +} diff --git a/src/openhuman/memory/tree/schemas.rs b/src/openhuman/memory/tree/schemas.rs new file mode 100644 index 000000000..5ce660f45 --- /dev/null +++ b/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 { + vec![ + schemas("ingest"), + schemas("list_chunks"), + schemas("get_chunk"), + ] +} + +pub fn all_registered_controllers() -> Vec { + 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) -> ControllerFuture { + Box::pin(async move { + let config = config_rpc::load_config_with_timeout().await?; + let req = parse_value::(Value::Object(params))?; + to_json(tree_rpc::ingest_rpc(&config, req).await?) + }) +} + +fn handle_list_chunks(params: Map) -> ControllerFuture { + Box::pin(async move { + let config = config_rpc::load_config_with_timeout().await?; + let req = parse_value::(Value::Object(params))?; + to_json(tree_rpc::list_chunks_rpc(&config, req).await?) + }) +} + +fn handle_get_chunk(params: Map) -> ControllerFuture { + Box::pin(async move { + let config = config_rpc::load_config_with_timeout().await?; + let req = parse_value::(Value::Object(params))?; + to_json(tree_rpc::get_chunk_rpc(&config, req).await?) + }) +} + +fn parse_value(v: Value) -> Result { + serde_json::from_value(v).map_err(|e| format!("invalid params: {e}")) +} + +fn to_json(outcome: RpcOutcome) -> Result { + outcome.into_cli_compatible_json() +} diff --git a/src/openhuman/memory/tree/store.rs b/src/openhuman/memory/tree/store.rs new file mode 100644 index 000000000..de19d6b36 --- /dev/null +++ b/src/openhuman/memory/tree/store.rs @@ -0,0 +1,414 @@ +//! SQLite-backed persistence for ingested chunks (Phase 1 / issue #707). +//! +//! The store lives at `/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 { + 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> { + 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, + pub source_id: Option, + pub owner: Option, + /// Inclusive lower bound on `timestamp` (milliseconds since epoch). + pub since_ms: Option, + /// Inclusive upper bound on `timestamp` (milliseconds since epoch). + pub until_ms: Option, + /// Max rows to return (default 100 when `None`). + pub limit: Option, +} + +/// List chunks matching the provided filters, ordered by `timestamp` DESC. +pub fn list_chunks(config: &Config, query: &ListChunksQuery) -> Result> { + 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> = 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::>>() + .context("Failed to collect chunks")?; + Ok(rows) + }) +} + +/// Count total chunks in the store (useful for tests / diagnostics). +pub fn count_chunks(config: &Config) -> Result { + 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 { + 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 = 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 = 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> { + 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(config: &Config, f: impl FnOnce(&Connection) -> Result) -> Result { + 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) -> 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::>(); + 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); + } +} diff --git a/src/openhuman/memory/tree/types.rs b/src/openhuman/memory/tree/types.rs new file mode 100644 index 000000000..2b880e3e3 --- /dev/null +++ b/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 { + 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 { + 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 (``) 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) -> 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, + /// 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, DateTime), + /// 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, + /// Opaque pointer back to the raw source record for drill-down / citation. + pub source_ref: Option, +} + +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, + owner: impl Into, + timestamp: DateTime, + ) -> 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, +} + +/// 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( + value: &(DateTime, DateTime), + serializer: S, + ) -> Result { + 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, DateTime), 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); + } +} diff --git a/tests/json_rpc_e2e.rs b/tests/json_rpc_e2e.rs index 14f415e9a..b48830a06 100644 --- a/tests/json_rpc_e2e.rs +++ b/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();