feat(channels): expose WhatsApp Web data to agent via structured RPC API (#1308)

Co-authored-by: Cursor <cursoragent@cursor.com>

src/core/all.rs

@@ -49,6 +49,11 @@ impl RegisteredController {
 /// The global static registry of all controllers, initialized once on first access.
 static REGISTRY: OnceLock<Vec<RegisteredController>> = OnceLock::new();
 
+/// Internal-only controllers: registered for RPC dispatch but NOT in the agent-facing
+/// schema catalog.  These handlers are callable by trusted callers (e.g. the Tauri scanner)
+/// but should not be advertised to agents via tool listings or schema discovery.
+static INTERNAL_REGISTRY: OnceLock<Vec<RegisteredController>> = OnceLock::new();
+
 /// The global static registry of standalone CLI adapters.
 static CLI_ADAPTERS: OnceLock<Vec<RegisteredCliAdapter>> = OnceLock::new();
 
@@ -69,6 +74,16 @@ fn registry() -> &'static [RegisteredController] {
         .as_slice()
 }
 
+/// Returns a reference to the internal-only controller registry.
+///
+/// These controllers are callable over RPC but are NOT included in agent tool listings
+/// or schema discovery endpoints.
+fn internal_registry() -> &'static [RegisteredController] {
+    INTERNAL_REGISTRY
+        .get_or_init(build_internal_only_controllers)
+        .as_slice()
+}
+
 /// Returns a reference to the global CLI adapter registry.
 fn cli_adapters() -> &'static [RegisteredCliAdapter] {
     CLI_ADAPTERS.get_or_init(|| {
@@ -192,6 +207,21 @@ fn build_registered_controllers() -> Vec<RegisteredController> {
     );
     // Integration notification ingest, triage, and per-provider settings
     controllers.extend(crate::openhuman::notifications::all_notifications_registered_controllers());
+    // Structured WhatsApp Web data — agent-facing read-only controllers (list/search).
+    // The write-path ingest controller is registered separately in build_internal_only_controllers.
+    controllers.extend(crate::openhuman::whatsapp_data::all_whatsapp_data_registered_controllers());
+    controllers
+}
+
+/// Aggregates controllers that are registered for RPC routing but NOT exposed to agents.
+///
+/// These are write-path or internal-only handlers callable by trusted callers
+/// (e.g. the Tauri scanner ingest path) that should not appear in agent tool listings.
+fn build_internal_only_controllers() -> Vec<RegisteredController> {
+    let mut controllers = Vec::new();
+    // whatsapp_data ingest: scanner-side write path.  Callable over RPC by the
+    // Tauri scanner but excluded from agent-facing schema discovery.
+    controllers.extend(crate::openhuman::whatsapp_data::all_whatsapp_data_internal_controllers());
     controllers
 }
 
@@ -255,6 +285,8 @@ fn build_declared_controller_schemas() -> Vec<ControllerSchema> {
     );
     // Integration notification ingest, triage, and per-provider settings
     schemas.extend(crate::openhuman::notifications::all_notifications_controller_schemas());
+    // Structured WhatsApp Web data — local SQLite store, agent-queryable
+    schemas.extend(crate::openhuman::whatsapp_data::all_whatsapp_data_controller_schemas());
     schemas
 }
 
@@ -340,6 +372,9 @@ pub fn namespace_description(namespace: &str) -> Option<&'static str> {
             "Integration notification ingest, triage scoring, listing, read-state, \
              and per-provider routing settings.",
         ),
+        "whatsapp_data" => Some(
+            "Structured WhatsApp conversation and message store — list chats, read messages, and search across WhatsApp Web data.",
+        ),
         _ => None,
     }
 }
@@ -361,9 +396,13 @@ pub fn rpc_method_from_parts(namespace: &str, function: &str) -> Option<String>
 }
 
 /// Retrieves the schema for a specific RPC method.
+///
+/// Checks both the agent-facing registry and the internal registry so that
+/// parameter validation still applies to internal-only methods (e.g. ingest).
 pub fn schema_for_rpc_method(method: &str) -> Option<ControllerSchema> {
     registry()
         .iter()
+        .chain(internal_registry().iter())
         .find(|r| r.rpc_method_name() == method)
         .map(|r| r.schema.clone())
 }
@@ -400,7 +439,11 @@ pub fn validate_params(
 
 /// Attempts to invoke a registered RPC method by name.
 ///
-/// Returns `None` if the method is not found in the registry.
+/// Checks both the agent-facing controller registry and the internal-only registry,
+/// so scanner-side write paths (e.g. `openhuman.whatsapp_data_ingest`) are routable
+/// even though they are not included in agent tool listings.
+///
+/// Returns `None` if the method is not found in either registry.
 pub async fn try_invoke_registered_rpc(
     method: &str,
     params: Map<String, Value>,
@@ -410,6 +453,11 @@ pub async fn try_invoke_registered_rpc(
             return Some((controller.handler)(params).await);
         }
     }
+    for controller in internal_registry() {
+        if controller.rpc_method_name() == method {
+            return Some((controller.handler)(params).await);
+        }
+    }
     None
 }
 

src/core/jsonrpc.rs

@@ -674,6 +674,15 @@ async fn run_server_inner(
             ),
             Err(e) => log::warn!("[boot] memory::global init failed: {e}"),
         }
+        // Initialize the WhatsApp data store so scanner ingest calls
+        // can write data without requiring a lazy-init fallback.
+        match crate::openhuman::whatsapp_data::global::init(cfg.workspace_dir.clone()) {
+            Ok(_) => log::info!(
+                "[boot] whatsapp_data::global initialized (workspace={})",
+                cfg.workspace_dir.display()
+            ),
+            Err(e) => log::warn!("[boot] whatsapp_data::global init failed: {e}"),
+        }
     }
 
     let (resolved_port, port_source) = match port {

src/openhuman/about_app/catalog.rs

@@ -686,6 +686,16 @@ const CAPABILITIES: &[Capability] = &[
         status: CapabilityStatus::Beta,
         privacy: None,
     },
+    Capability {
+        id: "channels.whatsapp_read_messages",
+        name: "Read WhatsApp Messages",
+        domain: "channels",
+        category: CapabilityCategory::Channels,
+        description: "Read and search WhatsApp Web conversations and messages after connecting WhatsApp in OpenHuman. Data is stored locally only and never transmitted.",
+        how_to: "Connect WhatsApp Web via Channels, then ask the agent to read or summarise your messages.",
+        status: CapabilityStatus::Beta,
+        privacy: LOCAL_RAW,
+    },
     Capability {
         id: "settings.configure_ai",
         name: "Configure AI",

src/openhuman/mod.rs

@@ -70,4 +70,5 @@ pub mod webhooks;
 pub mod webview_accounts;
 pub mod webview_apis;
 pub mod webview_notifications;
+pub mod whatsapp_data;
 pub mod workspace;

src/openhuman/whatsapp_data/global.rs

@@ -0,0 +1,57 @@
+//! Process-global WhatsApp data store singleton.
+//!
+//! One `WhatsAppDataStore` lives for the entire core process, shared by RPC
+//! handlers and any other subsystem that needs it.
+//!
+//! # Usage
+//!
+//! ```ignore
+//! // At startup:
+//! whatsapp_data::global::init(workspace_dir)?;
+//!
+//! // In RPC handlers:
+//! let store = whatsapp_data::global::store()?;
+//! ```
+
+use std::path::PathBuf;
+use std::sync::{Arc, OnceLock};
+
+use crate::openhuman::whatsapp_data::store::WhatsAppDataStore;
+
+/// Shared, thread-safe reference to the store.
+pub type WhatsAppDataStoreRef = Arc<WhatsAppDataStore>;
+
+static GLOBAL_STORE: OnceLock<WhatsAppDataStoreRef> = OnceLock::new();
+
+/// Initialise the global store from a workspace directory. Idempotent —
+/// only the first call has any effect; subsequent calls return the existing
+/// instance.
+pub fn init(workspace_dir: PathBuf) -> Result<WhatsAppDataStoreRef, String> {
+    if let Some(existing) = GLOBAL_STORE.get() {
+        log::debug!("[whatsapp_data:global] already initialised");
+        return Ok(Arc::clone(existing));
+    }
+    log::info!(
+        "[whatsapp_data:global] initialising store workspace={}",
+        workspace_dir.display()
+    );
+    let store = Arc::new(
+        WhatsAppDataStore::new(&workspace_dir)
+            .map_err(|e| format!("[whatsapp_data] store init failed: {e}"))?,
+    );
+    let _ = GLOBAL_STORE.set(Arc::clone(&store));
+    Ok(GLOBAL_STORE.get().cloned().unwrap_or(store))
+}
+
+/// Return the global store. Errors if [`init`] has not been called yet.
+pub fn store() -> Result<WhatsAppDataStoreRef, String> {
+    GLOBAL_STORE.get().cloned().ok_or_else(|| {
+        "whatsapp_data global store accessed before init — call init(workspace) at startup"
+            .to_string()
+    })
+}
+
+/// Return the global store if already initialised, without error.
+pub fn store_if_ready() -> Option<WhatsAppDataStoreRef> {
+    GLOBAL_STORE.get().cloned()
+}

src/openhuman/whatsapp_data/mod.rs

@@ -0,0 +1,29 @@
+//! Structured WhatsApp Web data — local-only SQLite persistence and agent API.
+//!
+//! This domain stores WhatsApp chats and messages scraped by the Tauri
+//! `whatsapp_scanner` via CDP, making them queryable by the agent through
+//! the JSON-RPC controller surface.
+//!
+//! **Data locality**: all data remains on-device in `whatsapp_data.db`; it is
+//! never transmitted to any external service.
+//!
+//! ## Agent-facing RPC methods (read-only)
+//! - `openhuman.whatsapp_data_list_chats`
+//! - `openhuman.whatsapp_data_list_messages`
+//! - `openhuman.whatsapp_data_search_messages`
+//!
+//! ## Internal-only RPC method (write, scanner-side)
+//! - `openhuman.whatsapp_data_ingest` — NOT exposed via agent tool listings
+
+pub mod global;
+pub mod ops;
+pub mod rpc;
+mod schemas;
+pub mod store;
+pub mod types;
+
+pub use schemas::{
+    all_controller_schemas as all_whatsapp_data_controller_schemas,
+    all_internal_controllers as all_whatsapp_data_internal_controllers,
+    all_registered_controllers as all_whatsapp_data_registered_controllers,
+};

src/openhuman/whatsapp_data/ops.rs

@@ -0,0 +1,200 @@
+//! Business logic for WhatsApp data ingestion and retrieval.
+//!
+//! All operations take a `&WhatsAppDataStore` so callers control the store
+//! lifetime (shared `Arc` at runtime, fresh instance in tests).
+
+use anyhow::Result;
+
+use crate::openhuman::whatsapp_data::{
+    store::WhatsAppDataStore,
+    types::{
+        IngestRequest, IngestResult, ListChatsRequest, ListMessagesRequest, SearchMessagesRequest,
+        WhatsAppChat, WhatsAppMessage,
+    },
+};
+
+/// Number of seconds in 90 days — the auto-prune horizon.
+const PRUNE_HORIZON_SECS: i64 = 90 * 24 * 60 * 60;
+
+/// Ingest a scanner snapshot: upsert chats and messages, then prune messages
+/// older than 90 days.
+///
+/// Returns counts for observability / logging at the RPC layer.
+pub fn ingest(store: &WhatsAppDataStore, req: IngestRequest) -> Result<IngestResult> {
+    log::debug!(
+        "[whatsapp_data] ingest start chats={} messages={} (account redacted)",
+        req.chats.len(),
+        req.messages.len()
+    );
+
+    let chats_upserted = store.upsert_chats(&req.account_id, &req.chats)?;
+    let messages_upserted = store.upsert_messages(&req.account_id, &req.messages)?;
+
+    let now = std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .map(|d| d.as_secs() as i64)
+        .unwrap_or(0);
+    let cutoff_ts = now - PRUNE_HORIZON_SECS;
+    let messages_pruned = store.prune_old_messages(cutoff_ts)?;
+
+    let result = IngestResult {
+        chats_upserted,
+        messages_upserted,
+        messages_pruned,
+    };
+    log::debug!(
+        "[whatsapp_data] ingest done chats_upserted={} messages_upserted={} pruned={} (account redacted)",
+        result.chats_upserted,
+        result.messages_upserted,
+        result.messages_pruned
+    );
+    Ok(result)
+}
+
+/// Return chats from the local store, optionally filtered by account.
+pub fn list_chats(store: &WhatsAppDataStore, req: ListChatsRequest) -> Result<Vec<WhatsAppChat>> {
+    log::debug!(
+        "[whatsapp_data] list_chats has_account={} limit={:?} offset={:?}",
+        req.account_id.is_some(),
+        req.limit,
+        req.offset
+    );
+    store.list_chats(&req)
+}
+
+/// Return messages for a chat, with optional time range and pagination.
+pub fn list_messages(
+    store: &WhatsAppDataStore,
+    req: ListMessagesRequest,
+) -> Result<Vec<WhatsAppMessage>> {
+    log::debug!(
+        "[whatsapp_data] list_messages has_account={} (chat/account redacted)",
+        req.account_id.is_some()
+    );
+    store.list_messages(&req)
+}
+
+/// Full-text search over message bodies.
+pub fn search_messages(
+    store: &WhatsAppDataStore,
+    req: SearchMessagesRequest,
+) -> Result<Vec<WhatsAppMessage>> {
+    log::debug!(
+        "[whatsapp_data] search_messages has_account={} has_chat={} (query/identifiers redacted)",
+        req.account_id.is_some(),
+        req.chat_id.is_some()
+    );
+    store.search_messages(&req)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::openhuman::whatsapp_data::types::{ChatMeta, IngestMessage};
+    use std::collections::HashMap;
+    use tempfile::tempdir;
+
+    fn make_store() -> (WhatsAppDataStore, tempfile::TempDir) {
+        let tmp = tempdir().expect("tempdir");
+        let store = WhatsAppDataStore::new(tmp.path()).expect("store");
+        (store, tmp)
+    }
+
+    fn sample_request() -> IngestRequest {
+        // Use a timestamp close to "now" so messages are not pruned by the
+        // 90-day auto-prune horizon.  We derive it from the system clock
+        // minus one hour so even on slow CI boxes the message is comfortably
+        // within the retention window.
+        let recent_ts = std::time::SystemTime::now()
+            .duration_since(std::time::UNIX_EPOCH)
+            .map(|d| d.as_secs() as i64 - 3600)
+            .unwrap_or(1_750_000_000);
+        let mut chats = HashMap::new();
+        chats.insert(
+            "alice@c.us".to_string(),
+            ChatMeta {
+                name: Some("Alice".to_string()),
+            },
+        );
+        IngestRequest {
+            account_id: "acct1".to_string(),
+            chats,
+            messages: vec![IngestMessage {
+                message_id: "msg1".to_string(),
+                chat_id: "alice@c.us".to_string(),
+                sender: Some("Alice".to_string()),
+                sender_jid: None,
+                from_me: Some(false),
+                body: Some("Hello!".to_string()),
+                timestamp: Some(recent_ts),
+                message_type: Some("chat".to_string()),
+                source: Some("cdp-dom".to_string()),
+            }],
+        }
+    }
+
+    #[test]
+    fn ingest_returns_correct_counts() {
+        let (store, _tmp) = make_store();
+        let result = ingest(&store, sample_request()).unwrap();
+        assert_eq!(result.chats_upserted, 1);
+        assert_eq!(result.messages_upserted, 1);
+    }
+
+    #[test]
+    fn list_chats_after_ingest() {
+        let (store, _tmp) = make_store();
+        ingest(&store, sample_request()).unwrap();
+
+        let chats = list_chats(
+            &store,
+            ListChatsRequest {
+                account_id: None,
+                limit: None,
+                offset: None,
+            },
+        )
+        .unwrap();
+        assert_eq!(chats.len(), 1);
+        assert_eq!(chats[0].chat_id, "alice@c.us");
+    }
+
+    #[test]
+    fn list_messages_after_ingest() {
+        let (store, _tmp) = make_store();
+        ingest(&store, sample_request()).unwrap();
+
+        let msgs = list_messages(
+            &store,
+            ListMessagesRequest {
+                chat_id: "alice@c.us".to_string(),
+                account_id: None,
+                since_ts: None,
+                until_ts: None,
+                limit: None,
+                offset: None,
+            },
+        )
+        .unwrap();
+        assert_eq!(msgs.len(), 1);
+        assert_eq!(msgs[0].body, "Hello!");
+    }
+
+    #[test]
+    fn search_messages_after_ingest() {
+        let (store, _tmp) = make_store();
+        ingest(&store, sample_request()).unwrap();
+
+        let results = search_messages(
+            &store,
+            SearchMessagesRequest {
+                query: "Hello".to_string(),
+                chat_id: None,
+                account_id: None,
+                limit: None,
+            },
+        )
+        .unwrap();
+        assert_eq!(results.len(), 1);
+    }
+}

src/openhuman/whatsapp_data/rpc.rs

@@ -0,0 +1,122 @@
+//! RPC handler functions for WhatsApp data domain.
+//!
+//! Each function:
+//!   1. Acquires the global `WhatsAppDataStore`.
+//!   2. Delegates to `ops::*` for business logic.
+//!   3. Returns an `RpcOutcome<T>`.
+//!
+//! When no WhatsApp session is active (store not yet initialised), the
+//! handlers return an actionable "not connected" error so the agent can
+//! surface a useful message instead of a crash.
+
+use anyhow::Result;
+
+use crate::openhuman::whatsapp_data::{
+    global, ops,
+    types::{
+        IngestRequest, IngestResult, ListChatsRequest, ListMessagesRequest, SearchMessagesRequest,
+        WhatsAppChat, WhatsAppMessage,
+    },
+};
+use crate::rpc::RpcOutcome;
+
+/// Ensure the global store is initialised.
+///
+/// On first call after core startup this may lazily initialise using the
+/// default workspace path. For the scanner-side ingest path the store is
+/// already warm from the `core_server` startup sequence.
+fn require_store() -> Result<global::WhatsAppDataStoreRef, String> {
+    global::store()
+}
+
+/// Ingest a WhatsApp scanner snapshot.
+///
+/// Called by the Tauri whatsapp_scanner after each full CDP scan tick.
+pub async fn whatsapp_data_ingest(req: IngestRequest) -> Result<RpcOutcome<IngestResult>, String> {
+    log::debug!(
+        "[whatsapp_data][rpc] ingest enter chats={} messages={} (account redacted)",
+        req.chats.len(),
+        req.messages.len()
+    );
+    let store = require_store()?;
+    let result = ops::ingest(&store, req).map_err(|e| {
+        log::warn!("[whatsapp_data][rpc] ingest error: {e}");
+        format!("[whatsapp_data] ingest failed: {e}")
+    })?;
+    log::debug!(
+        "[whatsapp_data][rpc] ingest ok chats={} messages={} pruned={}",
+        result.chats_upserted,
+        result.messages_upserted,
+        result.messages_pruned
+    );
+    Ok(RpcOutcome::single_log(
+        result,
+        "whatsapp_data ingest complete",
+    ))
+}
+
+/// List WhatsApp chats, optionally filtered by account.
+pub async fn whatsapp_data_list_chats(
+    req: ListChatsRequest,
+) -> Result<RpcOutcome<Vec<WhatsAppChat>>, String> {
+    log::debug!(
+        "[whatsapp_data][rpc] list_chats enter has_account={} limit={:?} offset={:?}",
+        req.account_id.is_some(),
+        req.limit,
+        req.offset
+    );
+    let store = require_store()?;
+    let chats = ops::list_chats(&store, req).map_err(|e| {
+        log::warn!("[whatsapp_data][rpc] list_chats error: {e}");
+        format!("[whatsapp_data] list_chats failed: {e}")
+    })?;
+    log::debug!("[whatsapp_data][rpc] list_chats ok count={}", chats.len());
+    Ok(RpcOutcome::single_log(
+        chats,
+        "whatsapp_data list_chats complete",
+    ))
+}
+
+/// List messages for a chat, with optional time range and pagination.
+pub async fn whatsapp_data_list_messages(
+    req: ListMessagesRequest,
+) -> Result<RpcOutcome<Vec<WhatsAppMessage>>, String> {
+    log::debug!(
+        "[whatsapp_data][rpc] list_messages enter has_account={} (chat redacted)",
+        req.account_id.is_some()
+    );
+    let store = require_store()?;
+    let msgs = ops::list_messages(&store, req).map_err(|e| {
+        log::warn!("[whatsapp_data][rpc] list_messages error: {e}");
+        format!("[whatsapp_data] list_messages failed: {e}")
+    })?;
+    log::debug!("[whatsapp_data][rpc] list_messages ok count={}", msgs.len());
+    Ok(RpcOutcome::single_log(
+        msgs,
+        "whatsapp_data list_messages complete",
+    ))
+}
+
+/// Full-text search over message bodies.
+pub async fn whatsapp_data_search_messages(
+    req: SearchMessagesRequest,
+) -> Result<RpcOutcome<Vec<WhatsAppMessage>>, String> {
+    log::debug!(
+        "[whatsapp_data][rpc] search_messages enter has_account={} has_chat={} (query/identifiers redacted)",
+        req.account_id.is_some(),
+        req.chat_id.is_some()
+    );
+    let store = require_store()?;
+    let results = ops::search_messages(&store, req).map_err(|e| {
+        log::warn!("[whatsapp_data][rpc] search_messages error: {e}");
+        format!("[whatsapp_data] search_messages failed: {e}")
+    })?;
+    log::debug!(
+        "[whatsapp_data][rpc] search_messages ok count={}",
+        results.len()
+    );
+    Ok(RpcOutcome::single_log(
+        results,
+        "whatsapp_data search_messages complete",
+    ))
+}

src/openhuman/whatsapp_data/schemas.rs

@@ -0,0 +1,263 @@
+//! Controller schemas and handler dispatch for the `whatsapp_data` namespace.
+//!
+//! Agent-facing (read-only) RPC methods:
+//!   - `openhuman.whatsapp_data_list_chats`
+//!   - `openhuman.whatsapp_data_list_messages`
+//!   - `openhuman.whatsapp_data_search_messages`
+//!
+//! Internal write path (NOT exposed to the agent controller registry):
+//!   - `openhuman.whatsapp_data_ingest` — called by the Tauri scanner only
+//!
+//! Keeping ingest off the agent-facing registry prevents an agent from
+//! mutating or poisoning the local WhatsApp store directly.
+
+use serde::de::DeserializeOwned;
+use serde_json::{Map, Value};
+
+use crate::core::all::{ControllerFuture, RegisteredController};
+use crate::core::{ControllerSchema, FieldSchema, TypeSchema};
+use crate::rpc::RpcOutcome;
+
+/// Returns controller schemas advertised to the agent (read-only subset).
+/// The ingest schema is intentionally excluded — it is an internal write path
+/// called by the scanner, not something the agent should be able to invoke.
+pub fn all_controller_schemas() -> Vec<ControllerSchema> {
+    vec![
+        schemas("list_chats"),
+        schemas("list_messages"),
+        schemas("search_messages"),
+    ]
+}
+
+/// Returns registered controllers for the agent-facing dispatcher (read-only).
+/// The ingest handler is registered separately via `all_internal_controllers()`
+/// and wired by the scanner — not through the agent controller registry.
+pub fn all_registered_controllers() -> Vec<RegisteredController> {
+    vec![
+        RegisteredController {
+            schema: schemas("list_chats"),
+            handler: handle_list_chats,
+        },
+        RegisteredController {
+            schema: schemas("list_messages"),
+            handler: handle_list_messages,
+        },
+        RegisteredController {
+            schema: schemas("search_messages"),
+            handler: handle_search_messages,
+        },
+    ]
+}
+
+/// Returns the full controller set including the internal ingest handler.
+/// Used by the core RPC dispatcher so the scanner can call
+/// `openhuman.whatsapp_data_ingest` over JSON-RPC without exposing it to agents.
+pub fn all_internal_controllers() -> Vec<RegisteredController> {
+    let mut controllers = all_registered_controllers();
+    controllers.insert(
+        0,
+        RegisteredController {
+            schema: schemas("ingest"),
+            handler: handle_ingest,
+        },
+    );
+    controllers
+}
+
+pub fn schemas(function: &str) -> ControllerSchema {
+    match function {
+        "ingest" => ControllerSchema {
+            namespace: "whatsapp_data",
+            function: "ingest",
+            description: "Ingest a WhatsApp Web scanner snapshot (chats + messages). \
+                          Called by the Tauri scanner after each CDP tick. Data is stored \
+                          locally only — never transmitted externally.",
+            inputs: vec![
+                required_string("account_id", "WhatsApp account identifier (phone JID)."),
+                FieldSchema {
+                    name: "chats",
+                    ty: TypeSchema::Json,
+                    comment: "Map of chat JID → {name: string | null}.",
+                    required: true,
+                },
+                FieldSchema {
+                    name: "messages",
+                    ty: TypeSchema::Array(Box::new(TypeSchema::Json)),
+                    comment: "Array of message objects to upsert.",
+                    required: true,
+                },
+            ],
+            outputs: vec![FieldSchema {
+                name: "result",
+                ty: TypeSchema::Json,
+                comment: "Ingest summary: chats_upserted, messages_upserted, messages_pruned.",
+                required: true,
+            }],
+        },
+        "list_chats" => ControllerSchema {
+            namespace: "whatsapp_data",
+            function: "list_chats",
+            description: "List locally-stored WhatsApp chats, ordered by most recent message. \
+                          When account_id is omitted, chats from all accounts are returned.",
+            inputs: vec![
+                optional_string("account_id", "Filter to a specific WhatsApp account."),
+                FieldSchema {
+                    name: "limit",
+                    ty: TypeSchema::Option(Box::new(TypeSchema::U64)),
+                    comment: "Maximum results (default 50).",
+                    required: false,
+                },
+                FieldSchema {
+                    name: "offset",
+                    ty: TypeSchema::Option(Box::new(TypeSchema::U64)),
+                    comment: "Pagination offset (default 0).",
+                    required: false,
+                },
+            ],
+            outputs: vec![FieldSchema {
+                name: "chats",
+                ty: TypeSchema::Array(Box::new(TypeSchema::Json)),
+                comment: "Array of WhatsAppChat objects.",
+                required: true,
+            }],
+        },
+        "list_messages" => ControllerSchema {
+            namespace: "whatsapp_data",
+            function: "list_messages",
+            description: "List messages for a WhatsApp chat, ordered by timestamp ascending. \
+                          When account_id is omitted, messages from all accounts are included.",
+            inputs: vec![
+                required_string("chat_id", "JID of the chat to retrieve messages for."),
+                optional_string("account_id", "Filter to a specific WhatsApp account."),
+                FieldSchema {
+                    name: "since_ts",
+                    ty: TypeSchema::Option(Box::new(TypeSchema::I64)),
+                    comment: "Only return messages at or after this Unix timestamp (seconds).",
+                    required: false,
+                },
+                FieldSchema {
+                    name: "until_ts",
+                    ty: TypeSchema::Option(Box::new(TypeSchema::I64)),
+                    comment: "Only return messages at or before this Unix timestamp (seconds).",
+                    required: false,
+                },
+                FieldSchema {
+                    name: "limit",
+                    ty: TypeSchema::Option(Box::new(TypeSchema::U64)),
+                    comment: "Maximum results (default 100).",
+                    required: false,
+                },
+                FieldSchema {
+                    name: "offset",
+                    ty: TypeSchema::Option(Box::new(TypeSchema::U64)),
+                    comment: "Pagination offset (default 0).",
+                    required: false,
+                },
+            ],
+            outputs: vec![FieldSchema {
+                name: "messages",
+                ty: TypeSchema::Array(Box::new(TypeSchema::Json)),
+                comment: "Array of WhatsAppMessage objects.",
+                required: true,
+            }],
+        },
+        "search_messages" => ControllerSchema {
+            namespace: "whatsapp_data",
+            function: "search_messages",
+            description:
+                "Search locally-stored WhatsApp message bodies. \
+                          Case-insensitive substring match. \
+                          When account_id / chat_id are omitted, all accounts / chats are searched.",
+            inputs: vec![
+                required_string("query", "Search query matched against message bodies."),
+                optional_string("chat_id", "Restrict search to a specific chat JID."),
+                optional_string(
+                    "account_id",
+                    "Restrict search to a specific WhatsApp account.",
+                ),
+                FieldSchema {
+                    name: "limit",
+                    ty: TypeSchema::Option(Box::new(TypeSchema::U64)),
+                    comment: "Maximum results (default 20).",
+                    required: false,
+                },
+            ],
+            outputs: vec![FieldSchema {
+                name: "messages",
+                ty: TypeSchema::Array(Box::new(TypeSchema::Json)),
+                comment: "Array of WhatsAppMessage objects matching the query.",
+                required: true,
+            }],
+        },
+        _ => ControllerSchema {
+            namespace: "whatsapp_data",
+            function: "unknown",
+            description: "Unknown whatsapp_data controller function.",
+            inputs: vec![],
+            outputs: vec![FieldSchema {
+                name: "error",
+                ty: TypeSchema::String,
+                comment: "Lookup error details.",
+                required: true,
+            }],
+        },
+    }
+}
+
+// ── Handlers ────────────────────────────────────────────────────────────────
+
+fn handle_ingest(params: Map<String, Value>) -> ControllerFuture {
+    Box::pin(async move {
+        let req = deserialize_params(params)?;
+        to_json(crate::openhuman::whatsapp_data::rpc::whatsapp_data_ingest(req).await?)
+    })
+}
+
+fn handle_list_chats(params: Map<String, Value>) -> ControllerFuture {
+    Box::pin(async move {
+        let req = deserialize_params(params)?;
+        to_json(crate::openhuman::whatsapp_data::rpc::whatsapp_data_list_chats(req).await?)
+    })
+}
+
+fn handle_list_messages(params: Map<String, Value>) -> ControllerFuture {
+    Box::pin(async move {
+        let req = deserialize_params(params)?;
+        to_json(crate::openhuman::whatsapp_data::rpc::whatsapp_data_list_messages(req).await?)
+    })
+}
+
+fn handle_search_messages(params: Map<String, Value>) -> ControllerFuture {
+    Box::pin(async move {
+        let req = deserialize_params(params)?;
+        to_json(crate::openhuman::whatsapp_data::rpc::whatsapp_data_search_messages(req).await?)
+    })
+}
+
+// ── Helpers ──────────────────────────────────────────────────────────────────
+
+fn deserialize_params<T: DeserializeOwned>(params: Map<String, Value>) -> Result<T, String> {
+    serde_json::from_value(Value::Object(params)).map_err(|e| format!("invalid params: {e}"))
+}
+
+fn to_json<T: serde::Serialize>(outcome: RpcOutcome<T>) -> Result<Value, String> {
+    outcome.into_cli_compatible_json()
+}
+
+fn required_string(name: &'static str, comment: &'static str) -> FieldSchema {
+    FieldSchema {
+        name,
+        ty: TypeSchema::String,
+        comment,
+        required: true,
+    }
+}
+
+fn optional_string(name: &'static str, comment: &'static str) -> FieldSchema {
+    FieldSchema {
+        name,
+        ty: TypeSchema::Option(Box::new(TypeSchema::String)),
+        comment,
+        required: false,
+    }
+}

src/openhuman/whatsapp_data/store.rs

@@ -0,0 +1,662 @@
+//! SQLite-backed persistence for structured WhatsApp Web data.
+//!
+//! Data is stored in a dedicated `whatsapp_data.db` file inside the
+//! workspace directory. Tables: `wa_chats` and `wa_messages`.
+//!
+//! This store is local-only; no data is transmitted to external services.
+
+use std::collections::HashMap;
+use std::path::Path;
+
+use anyhow::{Context, Result};
+use rusqlite::{params, Connection};
+
+use crate::openhuman::whatsapp_data::types::{
+    ChatMeta, IngestMessage, ListChatsRequest, ListMessagesRequest, SearchMessagesRequest,
+    WhatsAppChat, WhatsAppMessage,
+};
+
+/// SQLite-backed store for WhatsApp chats and messages.
+pub struct WhatsAppDataStore {
+    db_path: std::path::PathBuf,
+}
+
+impl WhatsAppDataStore {
+    /// Open or create the `whatsapp_data.db` SQLite database in `workspace_dir`.
+    /// The directory (and any parents) are created if they do not exist.
+    pub fn new(workspace_dir: &Path) -> Result<Self> {
+        let db_path = workspace_dir.join("whatsapp_data").join("whatsapp_data.db");
+        if let Some(parent) = db_path.parent() {
+            std::fs::create_dir_all(parent)
+                .with_context(|| format!("create whatsapp_data dir: {}", parent.display()))?;
+        }
+        log::debug!("[whatsapp_data] opening store at {}", db_path.display());
+        let store = Self { db_path };
+        store.init_schema()?;
+        Ok(store)
+    }
+
+    /// Initialize the schema. Idempotent — safe to call on every startup.
+    fn init_schema(&self) -> Result<()> {
+        let conn = self.open_conn()?;
+        conn.execute_batch(
+            "PRAGMA journal_mode = WAL;
+             PRAGMA foreign_keys = ON;
+
+             CREATE TABLE IF NOT EXISTS wa_chats (
+                 account_id      TEXT NOT NULL,
+                 chat_id         TEXT NOT NULL,
+                 display_name    TEXT NOT NULL DEFAULT '',
+                 is_group        INTEGER NOT NULL DEFAULT 0,
+                 last_message_ts INTEGER NOT NULL DEFAULT 0,
+                 message_count   INTEGER NOT NULL DEFAULT 0,
+                 updated_at      INTEGER NOT NULL DEFAULT 0,
+                 PRIMARY KEY (account_id, chat_id)
+             );
+
+             CREATE TABLE IF NOT EXISTS wa_messages (
+                 account_id   TEXT NOT NULL,
+                 chat_id      TEXT NOT NULL,
+                 message_id   TEXT NOT NULL,
+                 sender       TEXT NOT NULL DEFAULT '',
+                 sender_jid   TEXT,
+                 from_me      INTEGER NOT NULL DEFAULT 0,
+                 body         TEXT NOT NULL DEFAULT '',
+                 timestamp    INTEGER NOT NULL DEFAULT 0,
+                 message_type TEXT,
+                 source       TEXT NOT NULL DEFAULT '',
+                 ingested_at  INTEGER NOT NULL DEFAULT 0,
+                 PRIMARY KEY (account_id, chat_id, message_id)
+             );
+             CREATE INDEX IF NOT EXISTS idx_wa_msg_ts ON wa_messages(account_id, chat_id, timestamp);
+             CREATE INDEX IF NOT EXISTS idx_wa_msg_body ON wa_messages(account_id, body);",
+        )
+        .context("init whatsapp_data schema")?;
+        log::debug!("[whatsapp_data] schema ready");
+        Ok(())
+    }
+
+    fn open_conn(&self) -> Result<Connection> {
+        Connection::open(&self.db_path)
+            .with_context(|| format!("open whatsapp_data db: {}", self.db_path.display()))
+    }
+
+    fn now_secs() -> i64 {
+        std::time::SystemTime::now()
+            .duration_since(std::time::UNIX_EPOCH)
+            .map(|d| d.as_secs() as i64)
+            .unwrap_or(0)
+    }
+
+    /// Upsert chat metadata rows.  Returns the number of rows inserted or updated.
+    pub fn upsert_chats(
+        &self,
+        account_id: &str,
+        chats: &HashMap<String, ChatMeta>,
+    ) -> Result<usize> {
+        if chats.is_empty() {
+            return Ok(0);
+        }
+        let conn = self.open_conn()?;
+        let now = Self::now_secs();
+        let mut count = 0usize;
+        for (chat_id, meta) in chats {
+            let name = meta.name.as_deref().unwrap_or("");
+            let is_group = chat_id.ends_with("@g.us") as i64;
+            conn.execute(
+                "INSERT INTO wa_chats (account_id, chat_id, display_name, is_group, updated_at)
+                 VALUES (?1, ?2, ?3, ?4, ?5)
+                 ON CONFLICT(account_id, chat_id) DO UPDATE SET
+                     display_name = CASE WHEN excluded.display_name != '' THEN excluded.display_name ELSE display_name END,
+                     is_group     = excluded.is_group,
+                     updated_at   = excluded.updated_at",
+                params![account_id, chat_id, name, is_group, now],
+            )
+            .with_context(|| format!("upsert wa_chat {chat_id}"))?;
+            count += 1;
+        }
+        log::debug!(
+            "[whatsapp_data] upserted {} chats (account redacted)",
+            count
+        );
+        Ok(count)
+    }
+
+    /// Upsert message rows. Returns the number of rows inserted or updated.
+    pub fn upsert_messages(&self, account_id: &str, msgs: &[IngestMessage]) -> Result<usize> {
+        if msgs.is_empty() {
+            return Ok(0);
+        }
+        let conn = self.open_conn()?;
+        let now = Self::now_secs();
+        let mut count = 0usize;
+        for m in msgs {
+            if m.message_id.is_empty() || m.chat_id.is_empty() {
+                continue;
+            }
+            // Persist all messages, including non-text ones (stickers, images,
+            // system events).  Dropping empty-body rows biases message_count
+            // and last_message_ts to text-only messages, making active chats
+            // look stale whenever the latest event has no body.
+            let body = m.body.as_deref().unwrap_or("");
+            let ts = m.timestamp.unwrap_or(0);
+            let from_me = m.from_me.unwrap_or(false) as i64;
+            conn.execute(
+                "INSERT INTO wa_messages
+                     (account_id, chat_id, message_id, sender, sender_jid, from_me,
+                      body, timestamp, message_type, source, ingested_at)
+                 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11)
+                 ON CONFLICT(account_id, chat_id, message_id) DO UPDATE SET
+                     sender       = CASE WHEN excluded.sender != '' THEN excluded.sender ELSE sender END,
+                     sender_jid   = COALESCE(excluded.sender_jid, sender_jid),
+                     from_me      = excluded.from_me,
+                     body         = CASE WHEN excluded.body != '' THEN excluded.body ELSE body END,
+                     timestamp    = excluded.timestamp,
+                     message_type = COALESCE(excluded.message_type, message_type),
+                     source       = excluded.source,
+                     ingested_at  = excluded.ingested_at",
+                params![
+                    account_id,
+                    m.chat_id,
+                    m.message_id,
+                    m.sender.as_deref().unwrap_or(""),
+                    m.sender_jid.as_deref(),
+                    from_me,
+                    body,
+                    ts,
+                    m.message_type.as_deref(),
+                    m.source.as_deref().unwrap_or(""),
+                    now,
+                ],
+            )
+            .with_context(|| {
+                format!(
+                    "upsert wa_message chat={} msg={}",
+                    m.chat_id, m.message_id
+                )
+            })?;
+            count += 1;
+        }
+
+        // Refresh chat stats after message upsert.
+        if count > 0 {
+            conn.execute(
+                "UPDATE wa_chats
+                 SET message_count   = (SELECT COUNT(*) FROM wa_messages
+                                        WHERE wa_messages.account_id = wa_chats.account_id
+                                          AND wa_messages.chat_id    = wa_chats.chat_id),
+                     last_message_ts = COALESCE(
+                                         (SELECT MAX(timestamp) FROM wa_messages
+                                          WHERE wa_messages.account_id = wa_chats.account_id
+                                            AND wa_messages.chat_id    = wa_chats.chat_id),
+                                         last_message_ts),
+                     updated_at      = ?1
+                 WHERE account_id = ?2",
+                rusqlite::params![now, account_id],
+            )
+            .context("refresh wa_chats stats")?;
+        }
+
+        log::debug!(
+            "[whatsapp_data] upserted {} messages (account redacted)",
+            count
+        );
+        Ok(count)
+    }
+
+    /// Delete messages older than `cutoff_ts` (Unix seconds). Returns the count removed.
+    ///
+    /// After the delete, refreshes `wa_chats.message_count` and
+    /// `last_message_ts` for every chat that lost rows, so `list_chats`
+    /// returns accurate counts and ordering immediately.
+    pub fn prune_old_messages(&self, cutoff_ts: i64) -> Result<u64> {
+        let conn = self.open_conn()?;
+        let now = Self::now_secs();
+
+        // Collect affected (account_id, chat_id) pairs before deleting.
+        let mut stmt = conn.prepare(
+            "SELECT DISTINCT account_id, chat_id FROM wa_messages
+             WHERE timestamp > 0 AND timestamp < ?1",
+        )?;
+        let affected: Vec<(String, String)> = stmt
+            .query_map(params![cutoff_ts], |row| Ok((row.get(0)?, row.get(1)?)))?
+            .collect::<rusqlite::Result<_>>()
+            .context("collect affected chats for prune")?;
+
+        let changed = conn
+            .execute(
+                "DELETE FROM wa_messages WHERE timestamp > 0 AND timestamp < ?1",
+                params![cutoff_ts],
+            )
+            .context("prune old wa_messages")?;
+
+        // Refresh aggregate stats for every affected chat so list_chats
+        // reflects the post-prune state immediately.
+        if changed > 0 {
+            for (acct, chat_id) in &affected {
+                conn.execute(
+                    "UPDATE wa_chats
+                     SET message_count   = (SELECT COUNT(*) FROM wa_messages
+                                            WHERE account_id = wa_chats.account_id
+                                              AND chat_id    = wa_chats.chat_id),
+                         last_message_ts = COALESCE(
+                                             (SELECT MAX(timestamp) FROM wa_messages
+                                              WHERE account_id = wa_chats.account_id
+                                                AND chat_id    = wa_chats.chat_id),
+                                             last_message_ts),
+                         updated_at      = ?3
+                     WHERE account_id = ?1 AND chat_id = ?2",
+                    params![acct, chat_id, now],
+                )
+                .with_context(|| format!("refresh chat stats after prune: {chat_id}"))?;
+            }
+            log::debug!(
+                "[whatsapp_data] pruned {} messages (affected {} chats)",
+                changed,
+                affected.len()
+            );
+        }
+        Ok(changed as u64)
+    }
+
+    /// List chats, optionally filtered by account. Ordered by `last_message_ts` DESC.
+    pub fn list_chats(&self, req: &ListChatsRequest) -> Result<Vec<WhatsAppChat>> {
+        let conn = self.open_conn()?;
+        let limit = req.limit.unwrap_or(50) as i64;
+        let offset = req.offset.unwrap_or(0) as i64;
+
+        let chats = if let Some(ref acct) = req.account_id {
+            let mut stmt = conn.prepare(
+                "SELECT account_id, chat_id, display_name, is_group, last_message_ts,
+                        message_count, updated_at
+                 FROM wa_chats
+                 WHERE account_id = ?1
+                 ORDER BY last_message_ts DESC
+                 LIMIT ?2 OFFSET ?3",
+            )?;
+            let rows = stmt
+                .query_map(params![acct, limit, offset], map_chat_row)?
+                .collect::<rusqlite::Result<Vec<_>>>()
+                .context("list chats (filtered)")?;
+            rows
+        } else {
+            let mut stmt = conn.prepare(
+                "SELECT account_id, chat_id, display_name, is_group, last_message_ts,
+                        message_count, updated_at
+                 FROM wa_chats
+                 ORDER BY last_message_ts DESC
+                 LIMIT ?1 OFFSET ?2",
+            )?;
+            let rows = stmt
+                .query_map(params![limit, offset], map_chat_row)?
+                .collect::<rusqlite::Result<Vec<_>>>()
+                .context("list chats (all)")?;
+            rows
+        };
+        log::debug!("[whatsapp_data] list_chats returned {} rows", chats.len());
+        Ok(chats)
+    }
+
+    /// List messages for a chat, with optional time range and pagination.
+    pub fn list_messages(&self, req: &ListMessagesRequest) -> Result<Vec<WhatsAppMessage>> {
+        let conn = self.open_conn()?;
+        let limit = req.limit.unwrap_or(100) as i64;
+        let offset = req.offset.unwrap_or(0) as i64;
+        let since_ts = req.since_ts.unwrap_or(0);
+        let until_ts = req.until_ts.unwrap_or(i64::MAX);
+
+        let msgs = if let Some(ref acct) = req.account_id {
+            let mut stmt = conn.prepare(
+                "SELECT account_id, chat_id, message_id, sender, sender_jid, from_me,
+                        body, timestamp, message_type, source
+                 FROM wa_messages
+                 WHERE account_id = ?1
+                   AND chat_id    = ?2
+                   AND timestamp >= ?3
+                   AND timestamp <= ?4
+                 ORDER BY timestamp ASC
+                 LIMIT ?5 OFFSET ?6",
+            )?;
+            let rows = stmt
+                .query_map(
+                    params![acct, req.chat_id, since_ts, until_ts, limit, offset],
+                    map_message_row,
+                )?
+                .collect::<rusqlite::Result<Vec<_>>>()
+                .context("list messages (filtered by account)")?;
+            rows
+        } else {
+            let mut stmt = conn.prepare(
+                "SELECT account_id, chat_id, message_id, sender, sender_jid, from_me,
+                        body, timestamp, message_type, source
+                 FROM wa_messages
+                 WHERE chat_id    = ?1
+                   AND timestamp >= ?2
+                   AND timestamp <= ?3
+                 ORDER BY timestamp ASC
+                 LIMIT ?4 OFFSET ?5",
+            )?;
+            let rows = stmt
+                .query_map(
+                    params![req.chat_id, since_ts, until_ts, limit, offset],
+                    map_message_row,
+                )?
+                .collect::<rusqlite::Result<Vec<_>>>()
+                .context("list messages (all accounts)")?;
+            rows
+        };
+        log::debug!(
+            "[whatsapp_data] list_messages returned {} rows (chat/account redacted)",
+            msgs.len()
+        );
+        Ok(msgs)
+    }
+
+    /// Full-text search over message bodies (case-insensitive LIKE).
+    pub fn search_messages(&self, req: &SearchMessagesRequest) -> Result<Vec<WhatsAppMessage>> {
+        if req.query.trim().is_empty() {
+            return Ok(vec![]);
+        }
+        let conn = self.open_conn()?;
+        let limit = req.limit.unwrap_or(20) as i64;
+        let pattern = format!("%{}%", req.query.replace('%', "\\%").replace('_', "\\_"));
+
+        // Build the query dynamically depending on optional filters.
+        // Each branch binds to a local `rows` variable so `stmt` is dropped
+        // before the result is returned (fixes E0597 borrow lifetimes).
+        let msgs: Vec<WhatsAppMessage> = match (&req.account_id, &req.chat_id) {
+            (Some(acct), Some(chat_id)) => {
+                let mut stmt = conn.prepare(
+                    "SELECT account_id, chat_id, message_id, sender, sender_jid, from_me,
+                            body, timestamp, message_type, source
+                     FROM wa_messages
+                     WHERE account_id = ?1
+                       AND chat_id    = ?2
+                       AND body LIKE ?3 ESCAPE '\\'
+                     ORDER BY timestamp DESC
+                     LIMIT ?4",
+                )?;
+                let rows = stmt
+                    .query_map(params![acct, chat_id, pattern, limit], map_message_row)?
+                    .collect::<rusqlite::Result<Vec<_>>>()
+                    .context("search messages (account+chat)")?;
+                rows
+            }
+            (Some(acct), None) => {
+                let mut stmt = conn.prepare(
+                    "SELECT account_id, chat_id, message_id, sender, sender_jid, from_me,
+                            body, timestamp, message_type, source
+                     FROM wa_messages
+                     WHERE account_id = ?1
+                       AND body LIKE ?2 ESCAPE '\\'
+                     ORDER BY timestamp DESC
+                     LIMIT ?3",
+                )?;
+                let rows = stmt
+                    .query_map(params![acct, pattern, limit], map_message_row)?
+                    .collect::<rusqlite::Result<Vec<_>>>()
+                    .context("search messages (account)")?;
+                rows
+            }
+            (None, Some(chat_id)) => {
+                let mut stmt = conn.prepare(
+                    "SELECT account_id, chat_id, message_id, sender, sender_jid, from_me,
+                            body, timestamp, message_type, source
+                     FROM wa_messages
+                     WHERE chat_id = ?1
+                       AND body LIKE ?2 ESCAPE '\\'
+                     ORDER BY timestamp DESC
+                     LIMIT ?3",
+                )?;
+                let rows = stmt
+                    .query_map(params![chat_id, pattern, limit], map_message_row)?
+                    .collect::<rusqlite::Result<Vec<_>>>()
+                    .context("search messages (chat)")?;
+                rows
+            }
+            (None, None) => {
+                let mut stmt = conn.prepare(
+                    "SELECT account_id, chat_id, message_id, sender, sender_jid, from_me,
+                            body, timestamp, message_type, source
+                     FROM wa_messages
+                     WHERE body LIKE ?1 ESCAPE '\\'
+                     ORDER BY timestamp DESC
+                     LIMIT ?2",
+                )?;
+                let rows = stmt
+                    .query_map(params![pattern, limit], map_message_row)?
+                    .collect::<rusqlite::Result<Vec<_>>>()
+                    .context("search messages (all)")?;
+                rows
+            }
+        };
+        log::debug!(
+            "[whatsapp_data] search_messages returned {} rows (query/account redacted)",
+            msgs.len()
+        );
+        Ok(msgs)
+    }
+}
+
+fn map_chat_row(row: &rusqlite::Row<'_>) -> rusqlite::Result<WhatsAppChat> {
+    Ok(WhatsAppChat {
+        account_id: row.get(0)?,
+        chat_id: row.get(1)?,
+        display_name: row.get(2)?,
+        is_group: row.get::<_, i64>(3)? != 0,
+        last_message_ts: row.get(4)?,
+        message_count: row.get::<_, i64>(5)? as u32,
+        updated_at: row.get(6)?,
+    })
+}
+
+fn map_message_row(row: &rusqlite::Row<'_>) -> rusqlite::Result<WhatsAppMessage> {
+    Ok(WhatsAppMessage {
+        account_id: row.get(0)?,
+        chat_id: row.get(1)?,
+        message_id: row.get(2)?,
+        sender: row.get(3)?,
+        sender_jid: row.get(4)?,
+        from_me: row.get::<_, i64>(5)? != 0,
+        body: row.get(6)?,
+        timestamp: row.get(7)?,
+        message_type: row.get(8)?,
+        source: row.get(9)?,
+    })
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use tempfile::tempdir;
+
+    fn make_store() -> (WhatsAppDataStore, tempfile::TempDir) {
+        let tmp = tempdir().expect("tempdir");
+        let store = WhatsAppDataStore::new(tmp.path()).expect("store");
+        (store, tmp)
+    }
+
+    #[test]
+    fn upsert_and_list_chats() {
+        let (store, _tmp) = make_store();
+        let mut chats = HashMap::new();
+        chats.insert(
+            "chat1@c.us".to_string(),
+            ChatMeta {
+                name: Some("Alice".to_string()),
+            },
+        );
+        chats.insert(
+            "group1@g.us".to_string(),
+            ChatMeta {
+                name: Some("My Group".to_string()),
+            },
+        );
+        let count = store.upsert_chats("acct1", &chats).unwrap();
+        assert_eq!(count, 2);
+
+        let req = ListChatsRequest {
+            account_id: Some("acct1".to_string()),
+            limit: None,
+            offset: None,
+        };
+        let rows = store.list_chats(&req).unwrap();
+        assert_eq!(rows.len(), 2);
+
+        let group = rows.iter().find(|c| c.chat_id == "group1@g.us").unwrap();
+        assert!(group.is_group);
+        let dm = rows.iter().find(|c| c.chat_id == "chat1@c.us").unwrap();
+        assert!(!dm.is_group);
+    }
+
+    #[test]
+    fn upsert_and_list_messages() {
+        let (store, _tmp) = make_store();
+        let mut chats = HashMap::new();
+        chats.insert(
+            "chat1@c.us".to_string(),
+            ChatMeta {
+                name: Some("Alice".to_string()),
+            },
+        );
+        store.upsert_chats("acct1", &chats).unwrap();
+
+        let msgs = vec![
+            IngestMessage {
+                message_id: "msg1".to_string(),
+                chat_id: "chat1@c.us".to_string(),
+                sender: Some("Alice".to_string()),
+                sender_jid: None,
+                from_me: Some(false),
+                body: Some("Hello there".to_string()),
+                timestamp: Some(1_700_000_000),
+                message_type: Some("chat".to_string()),
+                source: Some("cdp-dom".to_string()),
+            },
+            IngestMessage {
+                message_id: "msg2".to_string(),
+                chat_id: "chat1@c.us".to_string(),
+                sender: Some("me".to_string()),
+                sender_jid: None,
+                from_me: Some(true),
+                body: Some("Hey!".to_string()),
+                timestamp: Some(1_700_000_100),
+                message_type: Some("chat".to_string()),
+                source: Some("cdp-indexeddb".to_string()),
+            },
+        ];
+        let count = store.upsert_messages("acct1", &msgs).unwrap();
+        assert_eq!(count, 2);
+
+        let req = ListMessagesRequest {
+            chat_id: "chat1@c.us".to_string(),
+            account_id: Some("acct1".to_string()),
+            since_ts: None,
+            until_ts: None,
+            limit: None,
+            offset: None,
+        };
+        let rows = store.list_messages(&req).unwrap();
+        assert_eq!(rows.len(), 2);
+        assert_eq!(rows[0].body, "Hello there");
+        assert_eq!(rows[1].body, "Hey!");
+    }
+
+    #[test]
+    fn search_messages_finds_match() {
+        let (store, _tmp) = make_store();
+        let mut chats = HashMap::new();
+        chats.insert(
+            "chat1@c.us".to_string(),
+            ChatMeta {
+                name: Some("Alice".to_string()),
+            },
+        );
+        store.upsert_chats("acct1", &chats).unwrap();
+
+        let msgs = vec![
+            IngestMessage {
+                message_id: "m1".to_string(),
+                chat_id: "chat1@c.us".to_string(),
+                sender: Some("Alice".to_string()),
+                sender_jid: None,
+                from_me: Some(false),
+                body: Some("Can you bring the umbrella?".to_string()),
+                timestamp: Some(1_700_000_000),
+                message_type: None,
+                source: Some("cdp-dom".to_string()),
+            },
+            IngestMessage {
+                message_id: "m2".to_string(),
+                chat_id: "chat1@c.us".to_string(),
+                sender: Some("me".to_string()),
+                sender_jid: None,
+                from_me: Some(true),
+                body: Some("Sure, no problem".to_string()),
+                timestamp: Some(1_700_000_200),
+                message_type: None,
+                source: Some("cdp-dom".to_string()),
+            },
+        ];
+        store.upsert_messages("acct1", &msgs).unwrap();
+
+        let req = SearchMessagesRequest {
+            query: "umbrella".to_string(),
+            chat_id: None,
+            account_id: None,
+            limit: None,
+        };
+        let results = store.search_messages(&req).unwrap();
+        assert_eq!(results.len(), 1);
+        assert!(results[0].body.contains("umbrella"));
+    }
+
+    #[test]
+    fn prune_removes_old_messages() {
+        let (store, _tmp) = make_store();
+        let mut chats = HashMap::new();
+        chats.insert("chat1@c.us".to_string(), ChatMeta { name: None });
+        store.upsert_chats("acct1", &chats).unwrap();
+
+        let msgs = vec![
+            IngestMessage {
+                message_id: "old".to_string(),
+                chat_id: "chat1@c.us".to_string(),
+                sender: None,
+                sender_jid: None,
+                from_me: Some(false),
+                body: Some("Old message".to_string()),
+                timestamp: Some(1_000_000),
+                message_type: None,
+                source: None,
+            },
+            IngestMessage {
+                message_id: "new".to_string(),
+                chat_id: "chat1@c.us".to_string(),
+                sender: None,
+                sender_jid: None,
+                from_me: Some(false),
+                body: Some("New message".to_string()),
+                timestamp: Some(2_000_000_000),
+                message_type: None,
+                source: None,
+            },
+        ];
+        store.upsert_messages("acct1", &msgs).unwrap();
+
+        let pruned = store.prune_old_messages(1_500_000_000).unwrap();
+        assert_eq!(pruned, 1);
+
+        let req = ListMessagesRequest {
+            chat_id: "chat1@c.us".to_string(),
+            account_id: None,
+            since_ts: None,
+            until_ts: None,
+            limit: None,
+            offset: None,
+        };
+        let remaining = store.list_messages(&req).unwrap();
+        assert_eq!(remaining.len(), 1);
+        assert_eq!(remaining[0].message_id, "new");
+    }
+}

src/openhuman/whatsapp_data/types.rs

@@ -0,0 +1,134 @@
+//! Normalized WhatsApp data structures — local-only, never transmitted externally.
+//!
+//! These types represent the structured data extracted from WhatsApp Web via CDP and
+//! persisted in a local SQLite database. All data remains local; nothing is sent to
+//! any remote service.
+
+use std::collections::HashMap;
+
+use serde::{Deserialize, Serialize};
+
+/// A WhatsApp chat (conversation) record stored locally.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct WhatsAppChat {
+    /// JID e.g. "123456@c.us" or "group@g.us"
+    pub chat_id: String,
+    /// Human-readable display name from WhatsApp contacts/group metadata.
+    pub display_name: String,
+    /// True if this chat is a group conversation.
+    pub is_group: bool,
+    /// The connected WhatsApp account identifier.
+    pub account_id: String,
+    /// Unix timestamp (seconds) of the most recent message stored.
+    pub last_message_ts: i64,
+    /// Number of messages stored for this chat.
+    pub message_count: u32,
+    /// Unix timestamp (seconds) when this record was last updated.
+    pub updated_at: i64,
+}
+
+/// A single WhatsApp message record stored locally.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct WhatsAppMessage {
+    /// WhatsApp message identifier (compound or bare form).
+    pub message_id: String,
+    /// JID of the chat this message belongs to.
+    pub chat_id: String,
+    /// Display name of the sender (stored as-is from WhatsApp).
+    pub sender: String,
+    /// JID of the sender, when available from IDB metadata.
+    pub sender_jid: Option<String>,
+    /// True if the message was sent by the account owner.
+    pub from_me: bool,
+    /// Decrypted message body text.
+    pub body: String,
+    /// Unix timestamp (seconds) of the message.
+    pub timestamp: i64,
+    /// WhatsApp message type (e.g. "chat", "image", "sticker").
+    pub message_type: Option<String>,
+    /// The connected WhatsApp account identifier.
+    pub account_id: String,
+    /// Data source: "cdp-dom" or "cdp-indexeddb".
+    pub source: String,
+}
+
+/// Metadata about a single chat in an ingest payload.
+#[derive(Debug, Deserialize)]
+pub struct ChatMeta {
+    /// Display name for the chat, if available.
+    pub name: Option<String>,
+}
+
+/// A single message entry in an ingest payload.
+#[derive(Debug, Deserialize)]
+pub struct IngestMessage {
+    pub message_id: String,
+    pub chat_id: String,
+    pub sender: Option<String>,
+    pub sender_jid: Option<String>,
+    pub from_me: Option<bool>,
+    pub body: Option<String>,
+    pub timestamp: Option<i64>,
+    pub message_type: Option<String>,
+    pub source: Option<String>,
+}
+
+/// Request payload for `openhuman.whatsapp_data_ingest`.
+#[derive(Debug, Deserialize)]
+pub struct IngestRequest {
+    /// The WhatsApp account identifier (usually the phone JID).
+    pub account_id: String,
+    /// Map of chat JID → chat metadata (display name, etc.).
+    pub chats: HashMap<String, ChatMeta>,
+    /// Messages to upsert into the local store.
+    pub messages: Vec<IngestMessage>,
+}
+
+/// Summary result returned after an ingest operation.
+#[derive(Debug, Serialize)]
+pub struct IngestResult {
+    pub chats_upserted: usize,
+    pub messages_upserted: usize,
+    pub messages_pruned: u64,
+}
+
+/// Request payload for `openhuman.whatsapp_data_list_chats`.
+#[derive(Debug, Deserialize)]
+pub struct ListChatsRequest {
+    /// Optional filter by account. When absent, all accounts are returned.
+    pub account_id: Option<String>,
+    /// Maximum number of results (default: 50).
+    pub limit: Option<u32>,
+    /// Pagination offset (default: 0).
+    pub offset: Option<u32>,
+}
+
+/// Request payload for `openhuman.whatsapp_data_list_messages`.
+#[derive(Debug, Deserialize)]
+pub struct ListMessagesRequest {
+    /// JID of the chat to retrieve messages for.
+    pub chat_id: String,
+    /// Optional filter by account. When absent, all accounts are searched.
+    pub account_id: Option<String>,
+    /// Only return messages at or after this Unix timestamp (seconds).
+    pub since_ts: Option<i64>,
+    /// Only return messages at or before this Unix timestamp (seconds).
+    pub until_ts: Option<i64>,
+    /// Maximum number of results (default: 100).
+    pub limit: Option<u32>,
+    /// Pagination offset (default: 0).
+    pub offset: Option<u32>,
+}
+
+/// Request payload for `openhuman.whatsapp_data_search_messages`.
+#[derive(Debug, Deserialize)]
+pub struct SearchMessagesRequest {
+    /// Full-text search query matched against message bodies (case-insensitive LIKE).
+    pub query: String,
+    /// Optional filter by chat JID.
+    pub chat_id: Option<String>,
+    /// Optional filter by account. When absent, all accounts are searched.
+    pub account_id: Option<String>,
+    /// Maximum number of results (default: 20).
+    pub limit: Option<u32>,
+}

tests/json_rpc_e2e.rs

@@ -3381,3 +3381,380 @@ async fn channels_status_reflects_managed_dm_credential_e2e() {
     mock_join.abort();
     rpc_join.abort();
 }
+
+/// WhatsApp data: ingest → list_chats → list_messages → search_messages
+///
+/// Validates the full structured data pipeline:
+///   1. Ingest two chats with five messages.
+///   2. list_chats returns both chats.
+///   3. list_messages for one chat returns the correct messages.
+///   4. search_messages finds the one matching message body.
+#[tokio::test]
+async fn whatsapp_data_ingest_and_query_e2e() {
+    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);
+
+    // Init the whatsapp_data global before the router handles any requests.
+    openhuman_core::openhuman::whatsapp_data::global::init(openhuman_home.clone())
+        .expect("whatsapp_data global init");
+
+    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;
+
+    // ── 1. Ingest: 2 chats, 5 messages ──────────────────────────────────────
+    // Use timestamps relative to now so the 90-day auto-prune never removes them.
+    let now_ts = chrono::Utc::now().timestamp();
+    let ingest = post_json_rpc(
+        &rpc_base,
+        9001,
+        "openhuman.whatsapp_data_ingest",
+        json!({
+            "account_id": "e2e-acct@c.us",
+            "chats": {
+                "alice@c.us": { "name": "Alice" },
+                "group1@g.us": { "name": "Friends Group" }
+            },
+            "messages": [
+                {
+                    "message_id": "msg-1",
+                    "chat_id": "alice@c.us",
+                    "sender": "Alice",
+                    "sender_jid": "alice@c.us",
+                    "from_me": false,
+                    "body": "Hey, how are you?",
+                    "timestamp": now_ts - 3600,
+                    "message_type": "chat",
+                    "source": "cdp-dom"
+                },
+                {
+                    "message_id": "msg-2",
+                    "chat_id": "alice@c.us",
+                    "sender": "me",
+                    "sender_jid": null,
+                    "from_me": true,
+                    "body": "Doing great, thanks!",
+                    "timestamp": now_ts - 3540,
+                    "message_type": "chat",
+                    "source": "cdp-dom"
+                },
+                {
+                    "message_id": "msg-3",
+                    "chat_id": "alice@c.us",
+                    "sender": "Alice",
+                    "sender_jid": "alice@c.us",
+                    "from_me": false,
+                    "body": "Can you send me the umbrella report?",
+                    "timestamp": now_ts - 3480,
+                    "message_type": "chat",
+                    "source": "cdp-dom"
+                },
+                {
+                    "message_id": "msg-4",
+                    "chat_id": "group1@g.us",
+                    "sender": "Bob",
+                    "sender_jid": "bob@c.us",
+                    "from_me": false,
+                    "body": "Meeting rescheduled to 3pm",
+                    "timestamp": now_ts - 2600,
+                    "message_type": "chat",
+                    "source": "cdp-indexeddb"
+                },
+                {
+                    "message_id": "msg-5",
+                    "chat_id": "group1@g.us",
+                    "sender": "me",
+                    "sender_jid": null,
+                    "from_me": true,
+                    "body": "Got it, I'll be there",
+                    "timestamp": now_ts - 2540,
+                    "message_type": "chat",
+                    "source": "cdp-indexeddb"
+                }
+            ]
+        }),
+    )
+    .await;
+    let ingest_result = assert_no_jsonrpc_error(&ingest, "whatsapp_data_ingest");
+    // The result may be wrapped in a logs envelope {result: ..., logs: [...]}
+    // or returned bare depending on whether logs are present.
+    let ingest_inner = ingest_result.get("result").unwrap_or(ingest_result);
+    let chats_upserted = ingest_inner
+        .get("chats_upserted")
+        .and_then(Value::as_u64)
+        .unwrap_or_else(|| panic!("missing chats_upserted in: {ingest_result}"));
+    assert_eq!(
+        chats_upserted, 2,
+        "expected 2 chats upserted: {ingest_result}"
+    );
+
+    // ── 2. list_chats — both chats should appear ─────────────────────────────
+    let list_chats = post_json_rpc(
+        &rpc_base,
+        9002,
+        "openhuman.whatsapp_data_list_chats",
+        json!({ "account_id": "e2e-acct@c.us" }),
+    )
+    .await;
+    let list_chats_result = assert_no_jsonrpc_error(&list_chats, "whatsapp_data_list_chats");
+    // Unwrap the result/logs envelope if present, then find the chats array.
+    let list_chats_inner = list_chats_result.get("result").unwrap_or(list_chats_result);
+    let chats_arr = list_chats_inner
+        .as_array()
+        .or_else(|| list_chats_inner.get("chats").and_then(Value::as_array))
+        .unwrap_or_else(|| panic!("expected chats array: {list_chats_result}"));
+    assert_eq!(chats_arr.len(), 2, "expected 2 chats: {list_chats_result}");
+
+    let chat_ids: Vec<&str> = chats_arr
+        .iter()
+        .filter_map(|c| c.get("chat_id").and_then(Value::as_str))
+        .collect();
+    assert!(
+        chat_ids.contains(&"alice@c.us"),
+        "alice chat missing: {chat_ids:?}"
+    );
+    assert!(
+        chat_ids.contains(&"group1@g.us"),
+        "group chat missing: {chat_ids:?}"
+    );
+
+    // ── 3. list_messages — alice's chat should have 3 messages ───────────────
+    let list_msgs = post_json_rpc(
+        &rpc_base,
+        9003,
+        "openhuman.whatsapp_data_list_messages",
+        json!({
+            "chat_id": "alice@c.us",
+            "account_id": "e2e-acct@c.us"
+        }),
+    )
+    .await;
+    let list_msgs_result = assert_no_jsonrpc_error(&list_msgs, "whatsapp_data_list_messages");
+    let list_msgs_inner = list_msgs_result.get("result").unwrap_or(list_msgs_result);
+    let msgs_arr = list_msgs_inner
+        .as_array()
+        .or_else(|| list_msgs_inner.get("messages").and_then(Value::as_array))
+        .unwrap_or_else(|| panic!("expected messages array: {list_msgs_result}"));
+    assert_eq!(
+        msgs_arr.len(),
+        3,
+        "expected 3 messages for alice: {list_msgs_result}"
+    );
+
+    // Messages should be ordered by timestamp ascending.
+    let bodies: Vec<&str> = msgs_arr
+        .iter()
+        .filter_map(|m| m.get("body").and_then(Value::as_str))
+        .collect();
+    assert_eq!(bodies[0], "Hey, how are you?");
+    assert_eq!(bodies[1], "Doing great, thanks!");
+    assert_eq!(bodies[2], "Can you send me the umbrella report?");
+
+    // ── 4. search_messages — "umbrella" should match exactly 1 message ───────
+    let search = post_json_rpc(
+        &rpc_base,
+        9004,
+        "openhuman.whatsapp_data_search_messages",
+        json!({ "query": "umbrella" }),
+    )
+    .await;
+    let search_result = assert_no_jsonrpc_error(&search, "whatsapp_data_search_messages");
+    let search_inner = search_result.get("result").unwrap_or(search_result);
+    let search_arr = search_inner
+        .as_array()
+        .or_else(|| search_inner.get("messages").and_then(Value::as_array))
+        .unwrap_or_else(|| panic!("expected messages array from search: {search_result}"));
+    assert_eq!(
+        search_arr.len(),
+        1,
+        "expected exactly 1 message matching 'umbrella': {search_result}"
+    );
+    let found_body = search_arr[0]
+        .get("body")
+        .and_then(Value::as_str)
+        .unwrap_or("");
+    assert!(
+        found_body.contains("umbrella"),
+        "search result body should contain 'umbrella': {found_body}"
+    );
+
+    // ── 5. account isolation — search scoped to first account only ────────────
+    // Ingest a second account with a message that also contains "umbrella" to
+    // verify that account_id filtering prevents cross-account leakage.
+    let second_ingest = post_json_rpc(
+        &rpc_base,
+        9005,
+        "openhuman.whatsapp_data_ingest",
+        json!({
+            "account_id": "other-acct@c.us",
+            "chats": {
+                "contact@c.us": { "name": "Other Contact" }
+            },
+            "messages": [
+                {
+                    "message_id": "other-msg-1",
+                    "chat_id": "contact@c.us",
+                    "sender": "Other Contact",
+                    "sender_jid": "contact@c.us",
+                    "from_me": false,
+                    "body": "Can you bring the umbrella?",
+                    "timestamp": now_ts - 1000,
+                    "message_type": "chat",
+                    "source": "cdp-dom"
+                }
+            ]
+        }),
+    )
+    .await;
+    assert_no_jsonrpc_error(&second_ingest, "whatsapp_data_ingest (second account)");
+
+    // search scoped to first account should still return exactly 1 message and
+    // that message's account_id must be from the first account.
+    let scoped_search = post_json_rpc(
+        &rpc_base,
+        9006,
+        "openhuman.whatsapp_data_search_messages",
+        json!({
+            "query": "umbrella",
+            "account_id": "e2e-acct@c.us"
+        }),
+    )
+    .await;
+    let scoped_result =
+        assert_no_jsonrpc_error(&scoped_search, "whatsapp_data_search_messages (scoped)");
+    let scoped_inner = scoped_result.get("result").unwrap_or(scoped_result);
+    let scoped_arr = scoped_inner
+        .as_array()
+        .or_else(|| scoped_inner.get("messages").and_then(Value::as_array))
+        .unwrap_or_else(|| panic!("expected messages array from scoped search: {scoped_result}"));
+    assert_eq!(
+        scoped_arr.len(),
+        1,
+        "account-scoped search should return exactly 1 umbrella message: {scoped_result}"
+    );
+    // Every result must belong to the queried account.
+    for msg in scoped_arr {
+        let msg_acct = msg.get("account_id").and_then(Value::as_str).unwrap_or("");
+        assert_eq!(
+            msg_acct, "e2e-acct@c.us",
+            "scoped search returned message from wrong account: {msg}"
+        );
+    }
+
+    mock_join.abort();
+    rpc_join.abort();
+}
+
+#[tokio::test]
+async fn whatsapp_memory_doc_ingest_e2e() {
+    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");
+    // Disable strict embedding so ingest falls back to the Inert
+    // (zero-vector) embedder when no Ollama endpoint is reachable. CI
+    // has no local Ollama; without this the memory_doc_ingest call
+    // would fail at the chunk-embedding step.
+    let _embed_strict_guard = EnvVarGuard::set("OPENHUMAN_MEMORY_EMBED_STRICT", "false");
+    let _embed_endpoint_guard = EnvVarGuard::set("OPENHUMAN_MEMORY_EMBED_ENDPOINT", "");
+    let _embed_model_guard = EnvVarGuard::set("OPENHUMAN_MEMORY_EMBED_MODEL", "");
+
+    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 (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;
+
+    // ── 1. Ingest a WhatsApp-shaped memory document ───────────────────────────
+    let ingest = post_json_rpc(
+        &rpc_base,
+        9101,
+        "openhuman.memory_doc_ingest",
+        json!({
+            "namespace": "whatsapp-web:test-acct@c.us",
+            "key": "alice@c.us:2026-05-07",
+            "title": "WhatsApp: Alice (2026-05-07)",
+            "content": "[10:00] Alice: Hey!\n[10:01] me: Hi there!\n[10:02] Alice: How are you?",
+            "source_type": "whatsapp-web",
+            "tags": ["whatsapp", "chat"],
+            "metadata": {
+                "chat_id": "alice@c.us",
+                "account_id": "test-acct@c.us"
+            }
+        }),
+    )
+    .await;
+    assert_no_jsonrpc_error(&ingest, "memory_doc_ingest");
+
+    // ── 2. List documents scoped to the WhatsApp namespace ───────────────────
+    let doc_list = post_json_rpc(
+        &rpc_base,
+        9102,
+        "openhuman.memory_doc_list",
+        json!({ "namespace": "whatsapp-web:test-acct@c.us" }),
+    )
+    .await;
+    let doc_list_result = assert_no_jsonrpc_error(&doc_list, "memory_doc_list");
+
+    // The result may be wrapped in a logs envelope {result: ..., logs: [...]}
+    // or returned bare depending on whether logs are present.
+    let doc_list_inner = doc_list_result.get("result").unwrap_or(doc_list_result);
+
+    // The doc_list response can be:
+    //   - an array directly
+    //   - { documents: [...], count: N }
+    //   - { result: [...] }
+    let docs_arr = doc_list_inner
+        .as_array()
+        .or_else(|| doc_list_inner.get("documents").and_then(Value::as_array))
+        .or_else(|| doc_list_inner.get("items").and_then(Value::as_array))
+        .unwrap_or_else(|| {
+            panic!("memory_doc_list: expected documents array in result: {doc_list_result}")
+        });
+
+    assert!(
+        !docs_arr.is_empty(),
+        "memory_doc_list should return at least 1 document after ingest: {doc_list_result}"
+    );
+
+    // ── 3. Verify the ingested document has the correct key and namespace ─────
+    let found = docs_arr.iter().find(|doc| {
+        let key_match = doc
+            .get("key")
+            .and_then(Value::as_str)
+            .map(|k| k == "alice@c.us:2026-05-07")
+            .unwrap_or(false);
+        let ns_match = doc
+            .get("namespace")
+            .and_then(Value::as_str)
+            .map(|n| n == "whatsapp-web:test-acct@c.us")
+            .unwrap_or(false);
+        key_match || ns_match
+    });
+    assert!(
+        found.is_some(),
+        "ingested document with key 'alice@c.us:2026-05-07' not found in doc_list; \
+         docs: {docs_arr:?}"
+    );
+
+    mock_join.abort();
+    rpc_join.abort();
+}