Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion docs/TEST-COVERAGE-MATRIX.md
Original file line number Diff line number Diff line change
Expand Up @@ -327,7 +327,8 @@ Canonical mapping of every product feature to its test source(s). Drives gap-fil
| ------ | ------------------------- | ----- | ----------------------------------------------------- | ------ | ----- |
| 10.3.1 | Incoming Message Sync | RU+WD | `src/openhuman/channels/tests/`, `gmail-flow.spec.ts` | ✅ | |
| 10.3.2 | Message Deduplication | RU | `src/openhuman/channels/tests/` | ✅ | |
| 10.3.3 | Real-Time vs Delayed Sync | RU | `src/openhuman/channels/tests/runtime_dispatch.rs` | ✅ | |
| 10.3.3 | WhatsApp Agent Retrieval | RU | `src/openhuman/tools/impl/whatsapp_data/` (this PR), `tests/json_rpc_e2e.rs::whatsapp_data_agent_tools_e2e_1341` (this PR) | ✅ | Three read-only agent tools wrap the local SQLite store; ingest stays internal-only. See [`docs/whatsapp-data-flow.md`](whatsapp-data-flow.md). |
| 10.3.4 | Real-Time vs Delayed Sync | RU | `src/openhuman/channels/tests/runtime_dispatch.rs` | ✅ | |

### 10.4 Messaging Operations

Expand Down
78 changes: 78 additions & 0 deletions docs/whatsapp-data-flow.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,78 @@
# WhatsApp data flow — scanner, store, agent

**Issue:** [#1341](https://github.com/tinyhumansai/openhuman/issues/1341)

This document describes how WhatsApp Web data captured by the desktop scanner becomes available to the agent. It exists to clear up the most common confusion: there are **two** local storage paths and they are intentional, not duplicates — each backs a different agent capability.

## Pipeline at a glance

```text
┌────────────────────────┐
│ WhatsApp Web (CEF view)│
└────────────┬───────────┘
│ CDP scan tick
▼
┌────────────────────────────────────┐
│ app/src-tauri/src/whatsapp_scanner │
│ (DOM + IndexedDB merge) │
└─────┬───────────────────────┬──────┘
│ exact rows │ canonicalised transcript
▼ ▼
┌──────────────────────┐ ┌──────────────────────────┐
│ openhuman.whatsapp_ │ │ openhuman.memory_doc_ │
│ data_ingest │ │ ingest │
│ (internal-only RPC) │ │ (internal-only RPC) │
└──────────┬───────────┘ └─────────────┬────────────┘
▼ ▼
┌──────────────────────┐ ┌──────────────────────────┐
│ whatsapp_data.db │ │ memory tree │
│ (SQLite, per-account)│ │ (per-source summaries + │
│ - wa_chats │ │ embeddings) │
│ - wa_messages │ │ │
└──────────┬───────────┘ └─────────────┬────────────┘
▼ ▼
┌──────────────────────┐ ┌──────────────────────────┐
│ Agent tools │ │ Agent tools │
│ whatsapp_data_* │ │ memory_tree_* │
│ (exact lookup) │ │ (semantic / cross-src) │
└──────────────────────┘ └──────────────────────────┘
```

Both ingest endpoints fire on every scan tick; both are `tokio::spawn` fire-and-forget so the scanner never blocks on either HTTP call.

## Why two paths

| Path | Backing store | Strength | Use it for |
|------|---------------|----------|------------|
| **Direct** | `whatsapp_data.db` (SQLite) | Exact, structured, paginated | "List my WhatsApp chats", "show the last 50 messages with Alice", "search for `invoice` across WhatsApp" |
| **Memory tree** | Per-source memory tree + embeddings | Semantic, cross-source | "Summarise this week of WhatsApp", "find action items across email and WhatsApp", "what did the team agree on?" |

The same scan tick populates both stores. Idempotency keys make the dual-write safe to retry:

- `whatsapp_data_ingest` keys on `(account_id, chat_id, message_id)` — UPSERT.
- `memory_doc_ingest` keys on `(namespace, key)` where namespace is `whatsapp-web:<account_id>` and key is `<chat_id>:<day>` — also UPSERT.

If one path fails (network blip, store init race), the other still progresses. The next scan tick converges both stores.

## Read-only boundary

The scanner write-path RPCs are registered as **internal-only** in [`src/core/all.rs`](../src/core/all.rs) under `build_internal_only_controllers`. They are reachable over JSON-RPC but invisible to the agent's tool catalog and to schema discovery (`all_controller_schemas`). The agent has **no** way to call `whatsapp_data_ingest` or `memory_doc_ingest` — accidentally or otherwise.

The agent surfaces are exclusively read-only:

- [`src/openhuman/tools/impl/whatsapp_data/`](../src/openhuman/tools/impl/whatsapp_data/) — `whatsapp_data_list_chats`, `whatsapp_data_list_messages`, `whatsapp_data_search_messages`. All three wrap their RPC counterparts and emit a `"provider": "whatsapp"` tag in the response so the agent can cite WhatsApp as the source.
- [`src/openhuman/tools/impl/memory/tree/`](../src/openhuman/tools/impl/memory/tree/) — generic `memory_tree_*` tools. Filter by `source_kind: "chat"` or query directly; WhatsApp chat-day transcripts are tagged `whatsapp` so they surface in cross-source flows.

## Why the orchestrator only lists three of these

The orchestrator's `agent.toml` exposes the three direct WhatsApp tools alongside the generic `memory_tree_*` family. That choice is deliberate — adding more provider-specific tools would compete with the memory-tree tools for the same intents and fragment routing. The combination satisfies the three real shapes of WhatsApp request:

1. **Exact lookup** ("what was my last message with Bob") → `whatsapp_data_list_messages` after `whatsapp_data_list_chats`.
2. **Keyword search** ("did anyone mention `Q3` on WhatsApp") → `whatsapp_data_search_messages`.
3. **Summarisation / action items / cross-source** ("what came up across WhatsApp and email this week") → `memory_tree_query_source { source_kind: "chat" }` or `memory_tree_query_global`.

If a future intent doesn't fit any of these, the right move is usually a new memory-tree retrieval primitive, not a new provider-specific tool.

## What this fix changed (#1341)

Prior to #1341 the read-only RPC controllers existed and were callable over JSON-RPC, but no `Tool` impl wrapped them and the orchestrator didn't list them — so the agent could see WhatsApp data only through the memory tree. That worked for summaries but failed on exact-lookup intents because the memory tree's per-day transcript granularity loses the structure the user asks about (sender JID, exact `chat_id`, per-message timestamp). Adding the three direct tools closed that gap without adding any new ingest path.
10 changes: 10 additions & 0 deletions src/openhuman/agent/agents/orchestrator/agent.toml
Original file line number Diff line number Diff line change
Expand Up @@ -80,6 +80,16 @@ named = [
"memory_tree_query_global",
"memory_tree_drill_down",
"memory_tree_fetch_leaves",
# WhatsApp local-data tools (issue #1341). The scanner ingests chats
# and messages into `whatsapp_data.db` on the user's machine; these
# three read-only tools let the orchestrator quote, summarise and
# search that local data without exposing the scanner's
# `whatsapp_data_ingest` write-path. Pair with the `memory_tree_*`
# tools above for cross-source / action-item flows once the scanner
# also forwards messages into the memory tree.
"whatsapp_data_list_chats",
"whatsapp_data_list_messages",
"whatsapp_data_search_messages",
"read_workspace_state",
"ask_user_clarification",
"spawn_subagent",
Expand Down
2 changes: 2 additions & 0 deletions src/openhuman/tools/impl/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@ pub mod filesystem;
pub mod memory;
pub mod network;
pub mod system;
pub mod whatsapp_data;

pub use agent::*;
pub use browser::*;
Expand All @@ -15,3 +16,4 @@ pub use filesystem::*;
pub use memory::*;
pub use network::*;
pub use system::*;
pub use whatsapp_data::*;
123 changes: 123 additions & 0 deletions src/openhuman/tools/impl/whatsapp_data/list_chats.rs
Original file line number Diff line number Diff line change
@@ -0,0 +1,123 @@
use crate::openhuman::tools::traits::{Tool, ToolResult};
use crate::openhuman::whatsapp_data::rpc as whatsapp_rpc;
use crate::openhuman::whatsapp_data::types::ListChatsRequest;
use async_trait::async_trait;
use serde_json::json;

pub struct WhatsAppDataListChatsTool;

#[async_trait]
impl Tool for WhatsAppDataListChatsTool {
fn name(&self) -> &str {
"whatsapp_data_list_chats"
}

fn description(&self) -> &str {
"List WhatsApp chats stored locally on this device, sorted by \
`last_message_ts` DESC (most recent activity first). USE THIS for \
intents about recent WhatsApp activity, identifying who the user \
spoke to recently, or resolving a contact/group name to a `chat_id`. \
Examples: 'who did I talk to on WhatsApp in the last 3 hours', \
'find my chat with Alice', 'which WhatsApp groups are active'. \
Each chat carries `chat_id`, `display_name`, `is_group`, \
`last_message_ts`, and `message_count`. After getting the chat \
list, drive `whatsapp_data_list_messages` (with the `chat_id`, \
optionally bounded by `since_ts`) for the message contents. \
Returns provider provenance so replies can cite WhatsApp."
}

fn parameters_schema(&self) -> serde_json::Value {
json!({
"type": "object",
"properties": {
"account_id": {
"type": "string",
"description": "Optional WhatsApp account JID. Omit to span every connected WhatsApp account."
},
"limit": {
"type": "integer",
"minimum": 1,
"description": "Maximum chats to return (default 50)."
},
"offset": {
"type": "integer",
"minimum": 0,
"description": "Pagination offset (default 0)."
}
}
})
}

async fn execute(&self, args: serde_json::Value) -> anyhow::Result<ToolResult> {
log::debug!("[tool][whatsapp_data] list_chats invoked");
let req: ListChatsRequest = serde_json::from_value(args).map_err(|e| {
log::debug!("[tool][whatsapp_data] list_chats invalid_args error={e}");
anyhow::anyhow!("invalid arguments for whatsapp_data_list_chats: {e}")
})?;
log::debug!(
"[tool][whatsapp_data] list_chats args has_account={} limit={:?} offset={:?}",
req.account_id.is_some(),
req.limit,
req.offset,
);
let outcome = whatsapp_rpc::whatsapp_data_list_chats(req)
.await
.map_err(|e| {
log::warn!("[tool][whatsapp_data] list_chats rpc_error error={e}");
anyhow::anyhow!("whatsapp_data_list_chats: {e}")
})?;
let chats = outcome.value;
log::debug!(
"[tool][whatsapp_data] list_chats returning count={}",
chats.len()
);
let body = serde_json::to_string(&json!({
"provider": "whatsapp",
"count": chats.len(),
"chats": chats,
}))?;
Ok(ToolResult::success(body))
}

fn is_concurrency_safe(&self, _args: &serde_json::Value) -> bool {
true
}
}

#[cfg(test)]
mod tests {
use super::*;
use crate::openhuman::tools::traits::{PermissionLevel, ToolScope};

#[test]
fn metadata_advertises_whatsapp() {
let tool = WhatsAppDataListChatsTool;
assert_eq!(tool.name(), "whatsapp_data_list_chats");
assert!(tool.description().contains("WhatsApp"));
assert_eq!(tool.permission_level(), PermissionLevel::ReadOnly);
assert_eq!(tool.scope(), ToolScope::All);
assert!(tool.is_concurrency_safe(&serde_json::Value::Null));
}

#[test]
fn parameters_schema_is_object_with_optional_fields() {
let schema = WhatsAppDataListChatsTool.parameters_schema();
assert_eq!(schema["type"], "object");
let props = &schema["properties"];
for key in ["account_id", "limit", "offset"] {
assert!(props.get(key).is_some(), "missing property {key}");
}
// No `required` array — every parameter is optional.
assert!(schema.get("required").is_none());
}

#[tokio::test]
async fn execute_rejects_invalid_args() {
let tool = WhatsAppDataListChatsTool;
let err = tool
.execute(json!({ "limit": "not-a-number" }))
.await
.expect_err("expected invalid-args error");
assert!(err.to_string().contains("whatsapp_data_list_chats"));
}
}
139 changes: 139 additions & 0 deletions src/openhuman/tools/impl/whatsapp_data/list_messages.rs
Original file line number Diff line number Diff line change
@@ -0,0 +1,139 @@
use crate::openhuman::tools::traits::{Tool, ToolResult};
use crate::openhuman::whatsapp_data::rpc as whatsapp_rpc;
use crate::openhuman::whatsapp_data::types::ListMessagesRequest;
use async_trait::async_trait;
use serde_json::json;

pub struct WhatsAppDataListMessagesTool;

#[async_trait]
impl Tool for WhatsAppDataListMessagesTool {
fn name(&self) -> &str {
"whatsapp_data_list_messages"
}

fn description(&self) -> &str {
"Return WhatsApp messages for one chat, ordered oldest-first within \
the requested time window. USE THIS for any WhatsApp request scoped \
to a specific chat — summarisation, action-item extraction, \
quoting, or 'show me the last N messages with <person>'. ALSO use \
this (paired with `since_ts` from `current_time` minus N hours) for \
time-window reads like 'what did <person> message me in the last 3 \
hours' AFTER resolving the chat via `whatsapp_data_list_chats`. The \
`chat_id` arg is required — get it from `whatsapp_data_list_chats` \
(or `whatsapp_data_search_messages`); optionally bound the range \
with `since_ts` / `until_ts` (Unix seconds). Do NOT use \
`whatsapp_data_search_messages` for time-only queries — that tool \
is keyword-based, not time-based. Returns provider provenance so \
replies can cite WhatsApp."
}

fn parameters_schema(&self) -> serde_json::Value {
json!({
"type": "object",
"properties": {
"chat_id": {
"type": "string",
"description": "WhatsApp chat JID, e.g. `1234567890@c.us` for a contact or `<id>@g.us` for a group. Required."
},
"account_id": {
"type": "string",
"description": "Optional WhatsApp account JID. Omit to span every connected account."
},
"since_ts": {
"type": "integer",
"description": "Lower bound (Unix seconds, inclusive) on message timestamp."
},
"until_ts": {
"type": "integer",
"description": "Upper bound (Unix seconds, inclusive) on message timestamp."
},
"limit": {
"type": "integer",
"minimum": 1,
"description": "Maximum messages to return (default 100)."
},
"offset": {
"type": "integer",
"minimum": 0,
"description": "Pagination offset (default 0)."
}
},
"required": ["chat_id"]
})
}

async fn execute(&self, args: serde_json::Value) -> anyhow::Result<ToolResult> {
log::debug!("[tool][whatsapp_data] list_messages invoked");
let req: ListMessagesRequest = serde_json::from_value(args).map_err(|e| {
log::debug!("[tool][whatsapp_data] list_messages invalid_args error={e}");
anyhow::anyhow!("invalid arguments for whatsapp_data_list_messages: {e}")
})?;
log::debug!(
"[tool][whatsapp_data] list_messages args has_account={} has_chat=true limit={:?} offset={:?} has_since={} has_until={}",
req.account_id.is_some(),
req.limit,
req.offset,
req.since_ts.is_some(),
req.until_ts.is_some(),
);
let outcome = whatsapp_rpc::whatsapp_data_list_messages(req)
.await
.map_err(|e| {
log::warn!("[tool][whatsapp_data] list_messages rpc_error error={e}");
anyhow::anyhow!("whatsapp_data_list_messages: {e}")
})?;
let messages = outcome.value;
log::debug!(
"[tool][whatsapp_data] list_messages returning count={}",
messages.len()
);
let body = serde_json::to_string(&json!({
"provider": "whatsapp",
"count": messages.len(),
"messages": messages,
}))?;
Ok(ToolResult::success(body))
}

fn is_concurrency_safe(&self, _args: &serde_json::Value) -> bool {
true
}
}

#[cfg(test)]
mod tests {
use super::*;
use crate::openhuman::tools::traits::{PermissionLevel, ToolScope};

#[test]
fn metadata_advertises_whatsapp() {
let tool = WhatsAppDataListMessagesTool;
assert_eq!(tool.name(), "whatsapp_data_list_messages");
assert!(tool.description().contains("WhatsApp"));
assert_eq!(tool.permission_level(), PermissionLevel::ReadOnly);
assert_eq!(tool.scope(), ToolScope::All);
assert!(tool.is_concurrency_safe(&serde_json::Value::Null));
}

#[test]
fn parameters_schema_requires_chat_id() {
let schema = WhatsAppDataListMessagesTool.parameters_schema();
assert_eq!(schema["type"], "object");
let required = schema["required"]
.as_array()
.expect("required array present");
let names: Vec<&str> = required.iter().filter_map(|v| v.as_str()).collect();
assert_eq!(names, vec!["chat_id"]);
}

#[tokio::test]
async fn execute_rejects_missing_chat_id() {
let tool = WhatsAppDataListMessagesTool;
let err = tool
.execute(json!({}))
.await
.expect_err("expected missing chat_id error");
assert!(err.to_string().contains("whatsapp_data_list_messages"));
}
}
Loading
Loading