diff --git a/src/memory/sync/composio/providers/common.rs b/src/memory/sync/composio/providers/common.rs index a0e5481..d50a1b0 100644 --- a/src/memory/sync/composio/providers/common.rs +++ b/src/memory/sync/composio/providers/common.rs @@ -4,17 +4,19 @@ use crate::memory::sync::traits::SkillDocument; /// Walk a JSON document by dotted path and return the first non-empty scalar. /// -/// # Not interchangeable with [`normalize::helpers::pick_str`] +/// # Not interchangeable with `tinymemory_sync::helpers::pick_str` /// -/// A second `pick_str` lives in [`normalize::helpers`], and the two differ. -/// This one resolves paths with [`Value::pointer`] (so a numeric segment -/// indexes into an array) and **coerces `Number` to its string form**; that -/// one walks with [`Value::get`] (objects only) and returns `None` for any +/// A second `pick_str` lives in the `tinymemory-sync` crate, and the two +/// differ. This one resolves paths with [`Value::pointer`] (so a numeric +/// segment indexes into an array) and **coerces `Number` to its string form**; +/// that one walks with [`Value::get`] (objects only) and returns `None` for any /// non-string leaf. Swapping one for the other changes what normalisers emit /// for numeric fields. Keep them separate. /// -/// [`normalize::helpers`]: super::normalize::helpers -/// [`normalize::helpers::pick_str`]: super::normalize::helpers::pick_str +/// The other one used to sit beside this file, under +/// `providers::normalize::helpers`. It moved out of this crate entirely with +/// the host-side normalisers (tinymemory#18 §B3), which is why this note names +/// a crate rather than a sibling module. pub fn pick_str(value: &Value, paths: &[&str]) -> Option { paths.iter().find_map(|path| { let pointer = format!("/{}", path.replace('.', "/")); diff --git a/src/memory/sync/composio/providers/mod.rs b/src/memory/sync/composio/providers/mod.rs index 2aa7fe7..7ca905b 100644 --- a/src/memory/sync/composio/providers/mod.rs +++ b/src/memory/sync/composio/providers/mod.rs @@ -8,7 +8,6 @@ mod google_docs; mod google_drive; mod google_sheets; mod linear; -pub mod normalize; mod notion; mod outlook; mod slack; diff --git a/src/memory/sync/composio/providers/normalize/clickup.rs b/src/memory/sync/composio/providers/normalize/clickup.rs deleted file mode 100644 index ae340b6..0000000 --- a/src/memory/sync/composio/providers/normalize/clickup.rs +++ /dev/null @@ -1,133 +0,0 @@ -//! ClickUp host normalization helpers — result extraction, task-title extraction, -//! and time utilities. -//! -//! ClickUp's REST API (and therefore Composio's wrapping of it) returns -//! task lists in a small handful of shapes depending on which endpoint -//! is called. The functions here walk the union of common shapes so the -//! provider doesn't have to branch per Composio envelope variant. - -use serde_json::Value; - -use super::helpers::pick_str; - -/// Walk the Composio response envelope for ClickUp task list results. -/// -/// ClickUp's "filtered team tasks" endpoint returns `{ "tasks": [...] }` -/// at the top level; Composio re-wraps the upstream payload under -/// `data` or `data.data` depending on the action. We probe each shape -/// in order and return the first array we find. -pub fn extract_tasks(data: &Value) -> Vec { - let candidates = [ - data.pointer("/data/tasks"), - data.pointer("/tasks"), - data.pointer("/data/data/tasks"), - data.pointer("/data/results"), - data.pointer("/results"), - data.pointer("/data/items"), - data.pointer("/items"), - ]; - for cand in candidates.into_iter().flatten() { - if let Some(arr) = cand.as_array() { - return arr.clone(); - } - } - Vec::new() -} - -/// Extract a human-readable title from a ClickUp task object. -/// -/// ClickUp tasks store the name at `name` (or `data.name` after Composio -/// envelope wrapping). When the name is missing we fall back to the -/// task ID so chunks remain identifiable. -pub fn extract_task_name(task: &Value) -> Option { - pick_str(task, &["name", "data.name", "title", "data.title"]) -} - -/// Extract a stable cursor timestamp (milliseconds since epoch as a -/// string) from a ClickUp task object. -/// -/// The ClickUp API returns `date_updated` as a stringified epoch ms -/// (e.g. `"1733412345678"`); we keep it as a string so lexicographic -/// comparison against the stored cursor remains valid as long as the -/// length doesn't change (it won't until year 33658). -pub fn extract_task_updated(task: &Value) -> Option { - pick_str( - task, - &[ - "date_updated", - "data.date_updated", - "updated_at", - "data.updated_at", - "dateUpdated", - "data.dateUpdated", - ], - ) -} - -/// Current wall-clock time in milliseconds since the UNIX epoch. -pub fn now_ms() -> u64 { - use std::time::{SystemTime, UNIX_EPOCH}; - SystemTime::now() - .duration_since(UNIX_EPOCH) - .map(|d| d.as_millis() as u64) - .unwrap_or(0) -} - -/// Extract the authorized user's numeric ID from the -/// `CLICKUP_GET_AUTHORIZED_USER` response. -/// -/// Composio wraps the upstream `{"user": {"id": …}}` shape; this walker -/// is defensive against both raw and wrapped payloads. Returns the ID -/// as a string because `CLICKUP_GET_FILTERED_TEAM_TASKS` accepts the -/// `assignees` filter as a string array. -pub fn extract_user_id(data: &Value) -> Option { - let candidates = [ - data.pointer("/user/id"), - data.pointer("/data/user/id"), - data.pointer("/id"), - data.pointer("/data/id"), - ]; - for cand in candidates.into_iter().flatten() { - if let Some(n) = cand.as_u64() { - return Some(n.to_string()); - } - if let Some(n) = cand.as_i64() { - return Some(n.to_string()); - } - if let Some(s) = cand.as_str() { - let trimmed = s.trim(); - if !trimmed.is_empty() { - return Some(trimmed.to_string()); - } - } - } - None -} - -/// Extract a list of workspace (team) IDs from the -/// `CLICKUP_GET_AUTHORIZED_TEAMS_WORKSPACES` response. -/// -/// ClickUp returns `{"teams": [{"id": "...", "name": "..."}, …]}`. We -/// keep the IDs as strings — `CLICKUP_GET_FILTERED_TEAM_TASKS` requires -/// a `team_id` (string) argument. -pub fn extract_workspace_ids(data: &Value) -> Vec { - let candidates = [ - data.pointer("/teams"), - data.pointer("/data/teams"), - data.pointer("/workspaces"), - data.pointer("/data/workspaces"), - ]; - for cand in candidates.into_iter().flatten() { - if let Some(arr) = cand.as_array() { - return arr - .iter() - .filter_map(|t| pick_str(t, &["id", "team_id", "workspace_id"])) - .collect(); - } - } - Vec::new() -} - -#[cfg(test)] -#[path = "clickup_tests.rs"] -mod tests; diff --git a/src/memory/sync/composio/providers/normalize/clickup_tests.rs b/src/memory/sync/composio/providers/normalize/clickup_tests.rs deleted file mode 100644 index 14ec117..0000000 --- a/src/memory/sync/composio/providers/normalize/clickup_tests.rs +++ /dev/null @@ -1,96 +0,0 @@ -use super::*; -use serde_json::json; - -#[test] -fn extract_tasks_from_data_tasks() { - let data = json!({ "data": { "tasks": [{"id": "t1"}] } }); - assert_eq!(extract_tasks(&data).len(), 1); -} - -#[test] -fn extract_tasks_from_top_level_tasks() { - let data = json!({ "tasks": [{"id": "a"}, {"id": "b"}] }); - assert_eq!(extract_tasks(&data).len(), 2); -} - -#[test] -fn extract_tasks_empty_when_missing() { - let data = json!({ "foo": "bar" }); - assert!(extract_tasks(&data).is_empty()); -} - -#[test] -fn extract_task_name_from_top_level() { - let task = json!({ "id": "t1", "name": "Build feature X" }); - assert_eq!(extract_task_name(&task), Some("Build feature X".into())); -} - -#[test] -fn extract_task_name_falls_back_to_data_name() { - let task = json!({ "data": { "name": "Wrapped" } }); - assert_eq!(extract_task_name(&task), Some("Wrapped".into())); -} - -#[test] -fn extract_task_name_none_when_missing() { - let task = json!({ "id": "t1" }); - assert!(extract_task_name(&task).is_none()); -} - -#[test] -fn extract_task_updated_handles_string_form() { - let task = json!({ "date_updated": "1733412345678" }); - assert_eq!( - extract_task_updated(&task), - Some("1733412345678".to_string()) - ); -} - -#[test] -fn extract_task_updated_handles_nested_data() { - let task = json!({ "data": { "dateUpdated": "1700000000000" } }); - assert_eq!( - extract_task_updated(&task), - Some("1700000000000".to_string()) - ); -} - -#[test] -fn extract_user_id_handles_numeric_id() { - let data = json!({ "user": { "id": 12345 } }); - assert_eq!(extract_user_id(&data), Some("12345".to_string())); -} - -#[test] -fn extract_user_id_handles_wrapped_payload() { - let data = json!({ "data": { "user": { "id": "777" } } }); - assert_eq!(extract_user_id(&data), Some("777".to_string())); -} - -#[test] -fn extract_user_id_none_when_missing() { - let data = json!({ "foo": "bar" }); - assert!(extract_user_id(&data).is_none()); -} - -#[test] -fn extract_workspace_ids_from_teams_array() { - let data = json!({ - "teams": [ - { "id": "ws1", "name": "Personal" }, - { "id": "ws2", "name": "Acme" }, - ] - }); - assert_eq!(extract_workspace_ids(&data), vec!["ws1", "ws2"]); -} - -#[test] -fn extract_workspace_ids_empty_when_no_teams() { - let data = json!({ "foo": "bar" }); - assert!(extract_workspace_ids(&data).is_empty()); -} - -#[test] -fn now_ms_returns_nonzero() { - assert!(now_ms() > 0); -} diff --git a/src/memory/sync/composio/providers/normalize/github.rs b/src/memory/sync/composio/providers/normalize/github.rs deleted file mode 100644 index 393e93e..0000000 --- a/src/memory/sync/composio/providers/normalize/github.rs +++ /dev/null @@ -1,130 +0,0 @@ -//! GitHub host normalization helpers — result extraction, identity helpers, and time utilities. -//! -//! GitHub's REST API (proxied through Composio) returns search results and -//! authenticated-user payloads in a small number of shapes. The functions here -//! walk the union of common Composio envelope variants so the provider stays -//! clean and branch-free. - -use serde_json::Value; - -use super::helpers::pick_str; - -/// Walk the Composio response envelope for GitHub search issue results. -/// -/// `GITHUB_SEARCH_ISSUES_AND_PULL_REQUESTS` wraps GitHub's `GET /search/issues` response, which -/// returns `{"total_count": N, "items": [...]}`. Composio may re-wrap this under -/// `data` or `data.data`; we probe each shape in order. -pub fn extract_issues(data: &Value) -> Vec { - let candidates = [ - data.pointer("/data/items"), - data.pointer("/items"), - data.pointer("/data/data/items"), - data.pointer("/data/results"), - data.pointer("/results"), - ]; - for cand in candidates.into_iter().flatten() { - if let Some(arr) = cand.as_array() { - return arr.clone(); - } - } - Vec::new() -} - -/// Extract a stable, globally unique identifier for a GitHub issue or PR. -/// -/// GitHub's internal `id` field is a large integer unique across all issues -/// and PRs on github.com. We convert it to a string for use as a sync key. -/// Falls back to composing from `html_url` path if `id` is absent. -pub fn extract_issue_id(issue: &Value) -> Option { - // Primary: numeric internal GitHub ID. - if let Some(id) = issue.get("id").or_else(|| issue.pointer("/data/id")) { - if let Some(n) = id.as_u64() { - return Some(n.to_string()); - } - if let Some(s) = id.as_str() { - let trimmed = s.trim(); - if !trimmed.is_empty() { - return Some(trimmed.to_string()); - } - } - } - // Fallback: parse owner/repo/number from html_url path segments. - // URL shape: https://github.com/{owner}/{repo}/issues/{number} - if let Some(url) = pick_str(issue, &["html_url", "data.html_url", "url", "data.url"]) { - if let Some(slug) = github_url_to_slug(&url) { - return Some(slug); - } - } - None -} - -/// Build a human-readable document title for a GitHub issue/PR. -/// -/// Format: `GitHub: {owner}/{repo}#{number}: {title}`. -/// Falls back to just the title or a placeholder when fields are missing. -pub fn extract_issue_title(issue: &Value) -> Option { - let title = pick_str(issue, &["title", "data.title"])?; - - // Best-effort: extract owner/repo#N from html_url for the prefix. - let prefix = pick_str(issue, &["html_url", "data.html_url"]) - .and_then(|url| github_url_to_slug(&url)) - .unwrap_or_default(); - - if prefix.is_empty() { - Some(title) - } else { - Some(format!("GitHub: {prefix}: {title}")) - } -} - -/// Parse `https://github.com/{owner}/{repo}/issues/{number}` (or `/pull/`) -/// into `"{owner}/{repo}#{number}"`. Returns `None` for unrecognised shapes. -fn github_url_to_slug(url: &str) -> Option { - let segs: Vec<&str> = url.trim_end_matches('/').split('/').collect(); - // Minimum: ["https:", "", "github.com", owner, repo, "issues", number] - if segs.len() >= 7 { - let number = segs[segs.len() - 1]; - let _kind = segs[segs.len() - 2]; // "issues" or "pull" — ignored - let repo = segs[segs.len() - 3]; - let owner = segs[segs.len() - 4]; - if !owner.is_empty() && !repo.is_empty() && !number.is_empty() { - return Some(format!("{owner}/{repo}#{number}")); - } - } - None -} - -/// Extract the `updated_at` ISO 8601 timestamp from a GitHub issue. -/// -/// GitHub returns `updated_at` as `"2024-05-21T15:30:00Z"`. ISO 8601 strings -/// sort lexicographically, so we use them directly as the sync cursor. -pub fn extract_issue_updated_at(issue: &Value) -> Option { - pick_str( - issue, - &[ - "updated_at", - "data.updated_at", - "updatedAt", - "data.updatedAt", - ], - ) -} - -/// Extract the authenticated user's login handle from a -/// `GITHUB_GET_THE_AUTHENTICATED_USER` response. -pub fn extract_user_login(data: &Value) -> Option { - pick_str(data, &["login", "data.login"]) -} - -/// Current wall-clock time in milliseconds since the UNIX epoch. -pub fn now_ms() -> u64 { - use std::time::{SystemTime, UNIX_EPOCH}; - SystemTime::now() - .duration_since(UNIX_EPOCH) - .map(|d| d.as_millis() as u64) - .unwrap_or(0) -} - -#[cfg(test)] -#[path = "github_tests.rs"] -mod tests; diff --git a/src/memory/sync/composio/providers/normalize/github_tests.rs b/src/memory/sync/composio/providers/normalize/github_tests.rs deleted file mode 100644 index 67cd3c5..0000000 --- a/src/memory/sync/composio/providers/normalize/github_tests.rs +++ /dev/null @@ -1,118 +0,0 @@ -use super::*; -use serde_json::json; - -#[test] -fn extract_issues_from_data_items() { - let data = json!({ "data": { "items": [{"id": 1}] } }); - assert_eq!(extract_issues(&data).len(), 1); -} - -#[test] -fn extract_issues_from_top_level_items() { - let data = json!({ "items": [{"id": 1}, {"id": 2}] }); - assert_eq!(extract_issues(&data).len(), 2); -} - -#[test] -fn extract_issues_empty_when_missing() { - let data = json!({ "foo": "bar" }); - assert!(extract_issues(&data).is_empty()); -} - -#[test] -fn extract_issue_id_from_numeric_field() { - let issue = json!({ "id": 123456789u64, "title": "Fix bug" }); - assert_eq!(extract_issue_id(&issue), Some("123456789".to_string())); -} - -#[test] -fn extract_issue_id_from_wrapped_data() { - let issue = json!({ "data": { "id": 99u64 } }); - assert_eq!(extract_issue_id(&issue), Some("99".to_string())); -} - -#[test] -fn extract_issue_id_falls_back_to_html_url() { - let issue = json!({ - "html_url": "https://github.com/owner/repo/issues/42" - }); - assert_eq!(extract_issue_id(&issue), Some("owner/repo#42".to_string())); -} - -#[test] -fn extract_issue_id_none_when_missing() { - let issue = json!({ "title": "No ID here" }); - assert!(extract_issue_id(&issue).is_none()); -} - -#[test] -fn extract_issue_title_builds_prefixed_title() { - let issue = json!({ - "id": 1u64, - "title": "Fix race condition", - "html_url": "https://github.com/acme/core/issues/99" - }); - assert_eq!( - extract_issue_title(&issue), - Some("GitHub: acme/core#99: Fix race condition".to_string()) - ); -} - -#[test] -fn extract_issue_title_returns_raw_title_when_no_url() { - let issue = json!({ "title": "Bare title" }); - assert_eq!(extract_issue_title(&issue), Some("Bare title".to_string())); -} - -#[test] -fn extract_issue_title_none_when_missing() { - let issue = json!({ "id": 1u64 }); - assert!(extract_issue_title(&issue).is_none()); -} - -#[test] -fn extract_issue_updated_at_from_top_level() { - let issue = json!({ "updated_at": "2024-05-21T15:30:00Z" }); - assert_eq!( - extract_issue_updated_at(&issue), - Some("2024-05-21T15:30:00Z".to_string()) - ); -} - -#[test] -fn extract_issue_updated_at_from_data_wrapper() { - let issue = json!({ "data": { "updated_at": "2023-01-01T00:00:00Z" } }); - assert_eq!( - extract_issue_updated_at(&issue), - Some("2023-01-01T00:00:00Z".to_string()) - ); -} - -#[test] -fn extract_issue_updated_at_none_when_missing() { - let issue = json!({ "id": 1u64 }); - assert!(extract_issue_updated_at(&issue).is_none()); -} - -#[test] -fn extract_user_login_from_top_level() { - let data = json!({ "login": "octocat" }); - assert_eq!(extract_user_login(&data), Some("octocat".to_string())); -} - -#[test] -fn extract_user_login_from_data_wrapper() { - let data = json!({ "data": { "login": "monalisa" } }); - assert_eq!(extract_user_login(&data), Some("monalisa".to_string())); -} - -#[test] -fn extract_user_login_none_when_missing() { - let data = json!({ "id": 1u64 }); - assert!(extract_user_login(&data).is_none()); -} - -#[test] -fn now_ms_returns_nonzero() { - assert!(now_ms() > 0); -} diff --git a/src/memory/sync/composio/providers/normalize/gmail_post_process.rs b/src/memory/sync/composio/providers/normalize/gmail_post_process.rs deleted file mode 100644 index 9404141..0000000 --- a/src/memory/sync/composio/providers/normalize/gmail_post_process.rs +++ /dev/null @@ -1,489 +0,0 @@ -//! Gmail-specific post-processing of Composio action responses. -//! -//! The upstream `GMAIL_FETCH_EMAILS` payload is extremely verbose -//! (full MIME tree under `payload.parts[]`, 50+ `Received:` headers, -//! display-layer noise the model never uses). This module rewrites -//! it into a slim envelope per message: -//! -//! ```json -//! { -//! "messages": [ -//! { -//! "id": "…", -//! "threadId": "…", -//! "subject": "…", -//! "from": "…", -//! "to": "…", -//! "date": "…", -//! "labels": ["INBOX", "UNREAD"], -//! "markdown": "…body…", -//! "attachments": [ { "filename": "...", "mimeType": "..." } ] -//! } -//! ], -//! "nextPageToken": "…", -//! "resultSizeEstimate": 201 -//! } -//! ``` -//! -//! ## Body source -//! -//! Composio's backend ships a -//! `markdownFormatted` field on the response envelope — one string -//! per tool call, pre-rendered with HTML stripped, URLs shortened, -//! footers removed, whitespace normalised. We split it per message -//! along `\n---\n` boundaries (with `## ` heading fallbacks) and -//! pin each slice to the corresponding entry in `messages[]` via -//! [`apply_response_level_markdown`]. The reshape's -//! `extract_markdown_body` then prefers that pinned field over -//! falling back to the upstream `messageText`. -//! -//! No in-house HTML→markdown conversion lives here anymore — the -//! backend does the cleaning. If `markdownFormatted` is absent for -//! a given response we fall through to whatever plain text the -//! upstream provided in `messageText`. -//! -//! Callers that need the raw Composio shape can pass `raw_html: -//! true` (or `rawHtml: true`) in the action arguments — this -//! short-circuits the reshape entirely. -//! -//! Only `GMAIL_FETCH_EMAILS` is reshaped today; other Gmail action -//! responses are passed through unchanged. When we add envelopes for -//! more slugs they should live in this file, branched from -//! [`post_process`]. - -use serde_json::{json, Map, Value}; - -/// Entry point called from `GmailProvider::post_process_action_result`. -/// -/// Dispatches on the Composio action slug. Unknown Gmail slugs fall -/// through to a no-op. -pub fn post_process(slug: &str, arguments: Option<&Value>, data: &mut Value) { - if is_raw_html_flag_set(arguments) { - tracing::debug!( - slug, - "[composio:gmail][post-process] raw_html flag set, passing through" - ); - return; - } - if slug == "GMAIL_FETCH_EMAILS" { - reshape_fetch_emails(data) - } -} - -/// Stash per-message slices of the response-level `markdownFormatted` -/// onto the corresponding entries inside `data.messages[]`. -/// -/// The Composio backend (tinyhumansai/backend#683) ships ONE -/// `markdownFormatted` string per tool call covering all messages — -/// already URL-shortened, footer-stripped, and whitespace-normalised. -/// To get per-email files in the raw archive we split that string -/// along section boundaries (`## ` headings or `---` rules) and pin -/// each slice to the message at the same index. `extract_markdown_body` -/// then prefers `msg.markdownFormatted` over re-decoding the MIME -/// tree. -/// -/// **Must be called BEFORE [`post_process`]** because `post_process` -/// reshapes `data` into the slim envelope; once `messages[]` carries -/// our slim shape the upstream message ordering is already locked in -/// but we may have lost original ordering signals if any. -/// -/// No-op when the slice count doesn't match `messages.len()` — we -/// can't safely align segments to messages without an exact match, -/// so we let `extract_markdown_body` fall through to its MIME path. -pub fn apply_response_level_markdown(data: &mut Value, top_md: &str) { - let trimmed = top_md.trim(); - if trimmed.is_empty() { - return; - } - let container = match data.get_mut("messages") { - Some(_) => data, - None => match data.get_mut("data").and_then(|v| v.as_object_mut()) { - Some(_) => data.get_mut("data").unwrap(), - None => { - tracing::debug!( - "[composio:gmail][post-process] apply_response_level_markdown: \ - no messages container in response — skipping" - ); - return; - } - }, - }; - let Some(messages) = container.get_mut("messages").and_then(|v| v.as_array_mut()) else { - return; - }; - let count = messages.len(); - if count == 0 { - return; - } - // Clone hints out of the messages array so the slice borrows - // don't conflict with the upcoming `messages.iter_mut()` mutation. - let hints: Vec = messages.clone(); - let Some(slices) = split_response_markdown_per_message_with_hint(trimmed, count, Some(&hints)) - else { - tracing::debug!( - messages = count, - md_len = trimmed.len(), - "[composio:gmail][post-process] could not split response-level markdownFormatted \ - into {count} slices — falling back to per-message MIME decode" - ); - return; - }; - for (msg, slice) in messages.iter_mut().zip(slices) { - if let Some(obj) = msg.as_object_mut() { - obj.insert("markdownFormatted".to_string(), Value::String(slice)); - } - } - tracing::debug!( - messages = count, - "[composio:gmail][post-process] stashed per-message markdownFormatted slices" - ); -} - -/// Split a top-level `markdownFormatted` string into per-message -/// segments. Returns `Some(slices)` only when the split yields -/// exactly `expected_count` entries — otherwise the format isn't one -/// of the patterns we know about and we let the caller fall back. -/// -/// Primary boundary is the `\n---\n` horizontal rule the backend -/// emits between messages (confirmed against real -/// `GMAIL_FETCH_EMAILS` output). H2/H3 headings are kept as -/// fallbacks for older renderings. The preamble (`# Inbox (N -/// messages)`-style intro, if present) is dropped — we accept -/// either `expected` parts (no preamble) or `expected + 1` -/// (preamble + N messages). -/// -/// `messages_hint` is the slim message array from the same response -/// — when present we use the per-message `subject` field to verify -/// each segment really does belong to the message at the same index. -/// Mismatches force a fallback so we never write a wrong-message body -/// to the raw archive. -pub fn split_response_markdown_per_message(md: &str, expected_count: usize) -> Option> { - split_response_markdown_per_message_with_hint(md, expected_count, None) -} - -pub fn split_response_markdown_per_message_with_hint( - md: &str, - expected_count: usize, - messages_hint: Option<&[Value]>, -) -> Option> { - if expected_count == 0 { - return None; - } - if expected_count == 1 { - return Some(vec![md.to_string()]); - } - - // Boundary patterns to try, in priority order. `\n---\n` is the - // confirmed marker; the heading variants stay as belt-and-braces - // for older / variant backend renderings. - let candidates: &[(&str, &str)] = &[ - ("\n---\n", "---\n"), - ("\n\n## ", "## "), - ("\n\n### ", "### "), - ("\n\n# ", "# "), - ("\n***\n", "***\n"), - ]; - - for (sep, prefix) in candidates { - let parts: Vec<&str> = md.split(sep).collect(); - let (drop_preamble, prepend_first) = if parts.len() == expected_count { - (false, false) // no preamble; first segment had no prefix - } else if parts.len() == expected_count + 1 { - (true, true) // preamble dropped; every kept segment had a prefix - } else { - continue; - }; - let segments: Vec = parts - .into_iter() - .skip(if drop_preamble { 1 } else { 0 }) - .enumerate() - .map(|(i, s)| { - if i == 0 && !prepend_first { - s.to_string() - } else { - format!("{prefix}{s}") - } - }) - .collect(); - - // Validate alignment against the JSON message array: every - // segment whose corresponding message has a non-empty subject - // must mention that subject somewhere in its body. If a single - // pair fails, we treat the split as unreliable and try the - // next pattern. Empty / null subjects skip validation (e.g. - // notification mails where the subject is ""). - if let Some(hints) = messages_hint { - if !validate_segments_against_hints(&segments, hints) { - tracing::debug!( - expected = expected_count, - sep = sep, - "[composio:gmail][post-process] split candidate failed subject check" - ); - continue; - } - } - return Some(segments); - } - None -} - -/// True if every (segment, message) pair where the message has a -/// non-empty subject contains that subject somewhere in the segment -/// (case-insensitive substring match — a defensive heuristic, not a -/// strict equality check, since the backend may format subjects -/// inside markdown links or with surrounding decoration). -fn validate_segments_against_hints(segments: &[String], hints: &[Value]) -> bool { - if segments.len() != hints.len() { - return false; - } - for (seg, hint) in segments.iter().zip(hints.iter()) { - let subject = hint - .get("subject") - .and_then(|v| v.as_str()) - .unwrap_or("") - .trim(); - if subject.is_empty() { - continue; - } - if !seg - .to_ascii_lowercase() - .contains(&subject.to_ascii_lowercase()) - { - return false; - } - } - true -} - -/// Returns true when the caller explicitly set `raw_html: true` (or the -/// camelCase `rawHtml: true`) in the `arguments` object. -fn is_raw_html_flag_set(arguments: Option<&Value>) -> bool { - let Some(obj) = arguments.and_then(|v| v.as_object()) else { - return false; - }; - obj.get("raw_html") - .or_else(|| obj.get("rawHtml")) - .and_then(|v| v.as_bool()) - .unwrap_or(false) -} - -/// Rewrite a `GMAIL_FETCH_EMAILS` `data` object in place into the slim -/// envelope documented at the module level. -/// -/// The Composio response can be shaped either as `{ messages, nextPageToken, ... }` -/// directly, or wrapped one level deeper under `{ data: { messages: … } }` -/// depending on backend version; we handle both. -fn reshape_fetch_emails(data: &mut Value) { - // Unwrap an optional `data:` envelope so downstream logic only has - // to deal with one shape. - let container = match data.get_mut("messages") { - Some(_) => data, - None => match data.get_mut("data").and_then(|v| v.as_object_mut()) { - Some(_) => data.get_mut("data").unwrap(), - None => return, - }, - }; - - let Some(obj) = container.as_object_mut() else { - return; - }; - - let raw_messages = obj - .remove("messages") - .and_then(|v| match v { - Value::Array(arr) => Some(arr), - _ => None, - }) - .unwrap_or_default(); - let next_page_token = obj.remove("nextPageToken").unwrap_or(Value::Null); - let result_size_estimate = obj.remove("resultSizeEstimate").unwrap_or(Value::Null); - - let messages: Vec = raw_messages.into_iter().map(reshape_message).collect(); - - let mut envelope = Map::new(); - envelope.insert("messages".into(), Value::Array(messages)); - if !next_page_token.is_null() { - envelope.insert("nextPageToken".into(), next_page_token); - } - if !result_size_estimate.is_null() { - envelope.insert("resultSizeEstimate".into(), result_size_estimate); - } - - *container = Value::Object(envelope); -} - -/// Parse an RFC 3339 or RFC 2822 date string into a UTC `DateTime`. -pub fn parse_email_date(date_str: &str) -> Option> { - date_str - .parse::>() - .or_else(|_| { - chrono::DateTime::parse_from_rfc2822(date_str).map(|d| d.with_timezone(&chrono::Utc)) - }) - .ok() -} - -const EMAIL_LOCAL_TIME_FMT: &str = "%Y-%m-%d %I:%M %p %:z"; - -/// Format a UTC `DateTime` in the given timezone. Returns `None` when the -/// formatted result is identical to the UTC rendering (no-op for UTC hosts). -pub fn format_at_tz( - utc: chrono::DateTime, - tz: &Tz, -) -> Option -where - Tz::Offset: std::fmt::Display, -{ - let local_dt = utc.with_timezone(tz); - let formatted = local_dt.format(EMAIL_LOCAL_TIME_FMT).to_string(); - - let utc_formatted = utc.format(EMAIL_LOCAL_TIME_FMT).to_string(); - if formatted == utc_formatted { - return None; - } - Some(formatted) -} - -/// Convert a UTC email timestamp string to a human-readable local-time string. -/// -/// Accepts RFC 3339 (`"2026-05-31T10:33:00Z"`) or RFC 2822 -/// (`"Sat, 31 May 2026 10:33:00 +0000"`) input. Returns a formatted string -/// in the host's local timezone, e.g. `"2026-05-31 05:33 AM -05:00"`, -/// so the agent can present local times without UTC arithmetic. -/// -/// The raw `date` field is always preserved alongside this field so -/// internal sorting, deduplication, and debugging remain UTC-based. -/// -/// Returns `None` when the input cannot be parsed or the output format -/// would be identical to the UTC input (no-op for UTC hosts). -pub fn format_email_local_time(date_str: &str) -> Option { - let utc = parse_email_date(date_str)?; - format_at_tz(utc, &chrono::Local) -} - -/// Map one raw Composio message object to its slim counterpart. -/// -/// Body source picked by [`extract_markdown_body`]: -/// 1. The per-message `markdownFormatted` slice pinned by -/// [`apply_response_level_markdown`] (preferred — backend-rendered). -/// 2. The upstream `messageText` plaintext (fallback). -/// 3. Empty string. -fn reshape_message(raw: Value) -> Value { - let Value::Object(obj) = raw else { - return raw; - }; - - let id = obj.get("messageId").cloned().unwrap_or(Value::Null); - let thread_id = obj.get("threadId").cloned().unwrap_or(Value::Null); - let subject = obj.get("subject").cloned().unwrap_or(Value::Null); - let sender = obj.get("sender").cloned().unwrap_or(Value::Null); - let to = obj.get("to").cloned().unwrap_or(Value::Null); - let date = obj - .get("messageTimestamp") - .cloned() - .or_else(|| pick_header(&obj, "Date")) - .unwrap_or(Value::Null); - let labels = obj - .get("labelIds") - .cloned() - .unwrap_or_else(|| Value::Array(Vec::new())); - let list_unsubscribe = pick_header(&obj, "List-Unsubscribe").unwrap_or(Value::Null); - - let markdown = extract_markdown_body(&obj); - let attachments = extract_attachments(&obj); - - // Compute a local-time representation of the UTC `date` so the agent - // presents times in the user's timezone rather than quoting raw UTC. - let date_local = date.as_str().and_then(format_email_local_time); - - let mut out = Map::new(); - out.insert("id".into(), id); - out.insert("threadId".into(), thread_id); - out.insert("subject".into(), subject); - out.insert("from".into(), sender); - out.insert("to".into(), to); - out.insert("date".into(), date); - if let Some(local) = date_local { - out.insert("date_local".into(), Value::String(local)); - } - out.insert("labels".into(), labels); - if !list_unsubscribe.is_null() { - out.insert("list_unsubscribe".into(), list_unsubscribe); - } - out.insert("markdown".into(), Value::String(markdown)); - if !attachments.is_empty() { - out.insert("attachments".into(), Value::Array(attachments)); - } - Value::Object(out) -} - -/// Find a header value by (case-insensitive) name in the Composio -/// `payload.headers[]` array. Returns `Some(Value::String)` on hit. -fn pick_header(msg: &Map, name: &str) -> Option { - let headers = msg.get("payload")?.get("headers")?.as_array()?; - for h in headers { - let hn = h.get("name").and_then(|v| v.as_str()).unwrap_or(""); - if hn.eq_ignore_ascii_case(name) { - if let Some(v) = h.get("value").and_then(|v| v.as_str()) { - return Some(Value::String(v.to_string())); - } - } - } - None -} - -/// Pick a body for the slim envelope. -/// -/// We trust the Composio backend's pre-rendered `markdownFormatted` -/// (set per-message by [`apply_response_level_markdown`] from the -/// response-level field). When that's absent we fall back to the -/// upstream's plain-text `messageText` verbatim — no in-house -/// HTML→markdown decoding lives here anymore. The backend already -/// strips HTML, shortens URLs, and normalises whitespace; running -/// our own pipeline on top duplicated work and corrupted some -/// renderings. -fn extract_markdown_body(msg: &Map) -> String { - if let Some(formatted) = msg - .get("markdownFormatted") - .or_else(|| msg.get("markdown_formatted")) - .and_then(|v| v.as_str()) - .map(str::trim) - .filter(|s| !s.is_empty()) - { - return formatted.to_string(); - } - if let Some(text) = msg - .get("messageText") - .and_then(|v| v.as_str()) - .map(str::trim) - .filter(|s| !s.is_empty()) - { - return text.to_string(); - } - String::new() -} - -/// Pull a minimal attachments descriptor from the Composio -/// `attachmentList` array. -fn extract_attachments(msg: &Map) -> Vec { - if let Some(list) = msg.get("attachmentList").and_then(|v| v.as_array()) { - return list - .iter() - .filter_map(|a| { - let filename = a.get("filename").and_then(|v| v.as_str())?; - if filename.is_empty() { - return None; - } - let mime = a - .get("mimeType") - .and_then(|v| v.as_str()) - .unwrap_or_default(); - Some(json!({ "filename": filename, "mimeType": mime })) - }) - .collect(); - } - Vec::new() -} - -#[cfg(test)] -#[path = "gmail_post_process_tests.rs"] -mod tests; diff --git a/src/memory/sync/composio/providers/normalize/gmail_post_process_tests.rs b/src/memory/sync/composio/providers/normalize/gmail_post_process_tests.rs deleted file mode 100644 index a143e95..0000000 --- a/src/memory/sync/composio/providers/normalize/gmail_post_process_tests.rs +++ /dev/null @@ -1,354 +0,0 @@ -use super::*; -use serde_json::json; - -fn fixture_with_backend_markdown() -> Value { - json!({ - "messages": [ - { - "messageId": "m1", - "threadId": "t1", - "subject": "Hello", - "sender": "a@x.com", - "to": "b@y.com", - "messageTimestamp": "2026-04-17T12:00:00Z", - "labelIds": ["INBOX", "UNREAD"], - // Pre-rendered slice (set by `apply_response_level_markdown` - // in production; inline here for the reshape test). - "markdownFormatted": "# Hello\n\nbody copy", - "messageText": "fallback should not be used", - "display_url": "ignore-me", - "preview": { "body": "Hi plain", "subject": "Hello" }, - "attachmentList": [ - { "filename": "report.pdf", "mimeType": "application/pdf", "size": 12345 }, - { "filename": "", "mimeType": "text/html" } - ], - "payload": {} - } - ], - "nextPageToken": "tok-1", - "resultSizeEstimate": 42 - }) -} - -#[test] -fn reshape_emits_slim_envelope() { - let mut v = fixture_with_backend_markdown(); - post_process("GMAIL_FETCH_EMAILS", None, &mut v); - - assert_eq!(v["nextPageToken"], "tok-1"); - assert_eq!(v["resultSizeEstimate"], 42); - - let msgs = v["messages"].as_array().unwrap(); - assert_eq!(msgs.len(), 1); - let m = &msgs[0]; - - assert_eq!(m["id"], "m1"); - assert_eq!(m["threadId"], "t1"); - assert_eq!(m["subject"], "Hello"); - assert_eq!(m["from"], "a@x.com"); - assert_eq!(m["to"], "b@y.com"); - assert_eq!(m["date"], "2026-04-17T12:00:00Z"); - assert_eq!(m["labels"], json!(["INBOX", "UNREAD"])); - - let md = m["markdown"].as_str().unwrap(); - assert_eq!(md, "# Hello\n\nbody copy"); - - // Noise fields removed. - assert!(m.get("display_url").is_none()); - assert!(m.get("preview").is_none()); - assert!(m.get("payload").is_none()); - assert!(m.get("messageText").is_none()); - - // Attachments: empty filename entry is filtered. - let atts = m["attachments"].as_array().unwrap(); - assert_eq!(atts.len(), 1); - assert_eq!(atts[0]["filename"], "report.pdf"); - assert_eq!(atts[0]["mimeType"], "application/pdf"); -} - -#[test] -fn raw_html_flag_passes_through_unchanged() { - let mut v = fixture_with_backend_markdown(); - let original = v.clone(); - let args = json!({ "raw_html": true }); - post_process("GMAIL_FETCH_EMAILS", Some(&args), &mut v); - assert_eq!( - v, original, - "raw_html=true must preserve the Composio shape" - ); -} - -#[test] -fn camel_case_raw_html_also_recognized() { - let mut v = fixture_with_backend_markdown(); - let original = v.clone(); - let args = json!({ "rawHtml": true }); - post_process("GMAIL_FETCH_EMAILS", Some(&args), &mut v); - assert_eq!(v, original); -} - -#[test] -fn falls_back_to_message_text_when_no_backend_markdown() { - let mut v = json!({ - "messages": [{ - "messageId": "m1", - "threadId": "t1", - "subject": "s", - "sender": "a@x.com", - "to": "b@y.com", - "messageTimestamp": "2026-04-17", - "labelIds": [], - "messageText": " plain body text ", - "payload": {} - }], - "nextPageToken": null - }); - post_process("GMAIL_FETCH_EMAILS", None, &mut v); - let md = v["messages"][0]["markdown"].as_str().unwrap(); - assert_eq!(md, "plain body text"); - assert!(v.get("nextPageToken").is_none(), "null tokens dropped"); -} - -#[test] -fn unwraps_data_envelope() { - let mut v = json!({ - "data": { - "messages": [{ - "messageId": "m1", - "threadId": "t1", - "subject": "s", - "sender": "a@x.com", - "to": "b@y.com", - "messageTimestamp": "2026-04-17", - "labelIds": [], - "messageText": "body", - "payload": {} - }] - } - }); - post_process("GMAIL_FETCH_EMAILS", None, &mut v); - // Reshape writes into `data` in place. - let msgs = v["data"]["messages"].as_array().unwrap(); - assert_eq!(msgs.len(), 1); - assert_eq!(msgs[0]["markdown"], "body"); -} - -#[test] -fn non_fetch_slug_is_noop() { - let mut v = json!({ "messages": [{ "messageId": "m1", "messageText": "x" }] }); - let original = v.clone(); - post_process("GMAIL_SEND_EMAIL", None, &mut v); - assert_eq!(v, original); -} - -#[test] -fn prefers_backend_markdown_formatted_when_present() { - // Composio backend (tinyhumansai/backend#683 +) ships - // `markdownFormatted` already URL-shortened + footer-stripped - // per message (after `apply_response_level_markdown` slices the - // response-level field). When present, our post-processor must - // use it verbatim instead of falling back to `messageText`. - let mut v = json!({ - "messages": [{ - "messageId": "m1", - "threadId": "t1", - "subject": "s", - "sender": "a@x.com", - "to": "b@y.com", - "messageTimestamp": "2026-04-17", - "labelIds": [], - "markdownFormatted": "# Already nice\n\nShort URL: https://gh.io/abc", - "messageText": "fallback should not be used", - "payload": {} - }] - }); - post_process("GMAIL_FETCH_EMAILS", None, &mut v); - let md = v["messages"][0]["markdown"].as_str().unwrap(); - assert_eq!(md, "# Already nice\n\nShort URL: https://gh.io/abc"); -} - -#[test] -fn empty_markdown_formatted_falls_through_to_message_text() { - let mut v = json!({ - "messages": [{ - "messageId": "m1", - "threadId": "t1", - "subject": "s", - "sender": "a@x.com", - "to": "b@y.com", - "messageTimestamp": "2026-04-17", - "labelIds": [], - "markdownFormatted": " \n \n", - "messageText": "real body", - "payload": {} - }] - }); - post_process("GMAIL_FETCH_EMAILS", None, &mut v); - let md = v["messages"][0]["markdown"].as_str().unwrap(); - assert!(md.contains("real body")); -} - -// ── split_response_markdown_per_message ───────────────────────────────── - -#[test] -fn split_response_markdown_uses_horizontal_rule_marker() { - // The confirmed backend marker is `\n---\n`. Three messages → - // expect three slices when there's no preamble. - let md = "## Alice's update\n\nbody A with https://gh.io/abc\n---\n## Bob's reply\n\nbody B\n---\n## Carol\n\nbody C"; - let slices = super::split_response_markdown_per_message(md, 3).unwrap(); - assert_eq!(slices.len(), 3); - assert!(slices[0].contains("Alice's update")); - assert!(slices[1].contains("Bob's reply")); - assert!(slices[2].contains("Carol")); - // The `---\n` prefix is preserved on every-but-the-first segment - // so the section break survives the round-trip. - assert!(slices[1].starts_with("---\n")); - assert!(slices[2].starts_with("---\n")); -} - -#[test] -fn split_response_markdown_drops_preamble() { - // When a preamble like `# Inbox` precedes the first marker, we - // see N+1 parts after split — the preamble must be dropped. - let md = "# Inbox (2 messages)\n---\n## A\n\nbody A\n---\n## B\n\nbody B"; - let slices = super::split_response_markdown_per_message(md, 2).unwrap(); - assert_eq!(slices.len(), 2); - assert!(slices[0].contains("body A")); - assert!(slices[1].contains("body B")); - // Both segments should carry the prefix when preamble was dropped. - assert!(slices[0].starts_with("---\n")); - assert!(slices[1].starts_with("---\n")); -} - -#[test] -fn split_response_markdown_falls_back_to_h2_marker() { - // No `---` rules — backend used h2 headings as boundaries. - let md = "## Alice\n\nbody A\n\n## Bob\n\nbody B"; - let slices = super::split_response_markdown_per_message(md, 2).unwrap(); - assert_eq!(slices.len(), 2); - assert!(slices[0].contains("body A")); - assert!(slices[1].contains("body B")); -} - -#[test] -fn split_response_markdown_returns_none_on_count_mismatch() { - let md = "## only one section here"; - assert!(super::split_response_markdown_per_message(md, 3).is_none()); -} - -#[test] -fn split_response_markdown_single_message_returns_whole_input() { - let md = "## solo\n\nthe whole body"; - let slices = super::split_response_markdown_per_message(md, 1).unwrap(); - assert_eq!(slices, vec![md.to_string()]); -} - -#[test] -fn split_with_hint_rejects_when_subjects_dont_match() { - let md = "## Foo\nbody1\n---\n## Bar\nbody2"; - let hints = vec![ - json!({"subject": "Completely different subject A"}), - json!({"subject": "Completely different subject B"}), - ]; - let out = super::split_response_markdown_per_message_with_hint(md, 2, Some(&hints)); - assert!(out.is_none(), "subject mismatch must force fallback"); -} - -#[test] -fn split_with_hint_accepts_when_subjects_match() { - let md = "## Welcome to Gmail\nbody1\n---\n## Your invoice\nbody2"; - let hints = vec![ - json!({"subject": "Welcome to Gmail"}), - json!({"subject": "Your invoice"}), - ]; - let slices = super::split_response_markdown_per_message_with_hint(md, 2, Some(&hints)).unwrap(); - assert_eq!(slices.len(), 2); - assert!(slices[0].contains("Welcome to Gmail")); - assert!(slices[1].contains("Your invoice")); -} - -#[test] -fn split_with_hint_skips_messages_with_blank_subject() { - let md = "## A\nbody1\n---\n## B\nbody2"; - let hints = vec![json!({"subject": "A"}), json!({"subject": ""})]; - let slices = super::split_response_markdown_per_message_with_hint(md, 2, Some(&hints)).unwrap(); - assert_eq!(slices.len(), 2); -} - -// ── format_email_local_time ────────────────────────────────────────────────── - -#[test] -fn format_email_local_time_returns_none_for_unparseable_date() { - assert!(super::format_email_local_time("not-a-date").is_none()); - assert!(super::format_email_local_time("").is_none()); -} - -#[test] -fn format_email_local_time_preserves_utc_raw_date_in_reshape() { - let mut v = json!({ - "messages": [{ - "messageId": "m1", - "threadId": "t1", - "subject": "Test", - "sender": "a@example.com", - "to": "b@example.com", - "messageTimestamp": "2026-05-31T10:33:00Z", - "labelIds": [], - "messageText": "body", - "payload": {} - }] - }); - post_process("GMAIL_FETCH_EMAILS", None, &mut v); - let msg = &v["messages"][0]; - assert_eq!(msg["date"], "2026-05-31T10:33:00Z"); -} - -#[test] -fn parse_email_date_accepts_rfc3339_and_rfc2822() { - assert!(super::parse_email_date("2026-05-31T10:33:00Z").is_some()); - assert!(super::parse_email_date("Sun, 31 May 2026 10:33:00 +0000").is_some()); - assert!(super::parse_email_date("not-a-date").is_none()); -} - -#[test] -fn format_at_tz_deterministic_with_fixed_offset() { - use chrono::FixedOffset; - - let utc = super::parse_email_date("2026-05-31T10:33:00Z").unwrap(); - - let est = FixedOffset::west_opt(5 * 3600).unwrap(); - let result = super::format_at_tz(utc, &est).unwrap(); - assert_eq!(result, "2026-05-31 05:33 AM -05:00"); - - let ist = FixedOffset::east_opt(5 * 3600 + 1800).unwrap(); - let result = super::format_at_tz(utc, &ist).unwrap(); - assert_eq!(result, "2026-05-31 04:03 PM +05:30"); -} - -#[test] -fn format_at_tz_returns_none_for_utc() { - let utc = super::parse_email_date("2026-05-31T10:33:00Z").unwrap(); - let utc_tz = chrono::FixedOffset::east_opt(0).unwrap(); - assert!(super::format_at_tz(utc, &utc_tz).is_none()); -} - -#[test] -fn apply_response_level_markdown_stashes_per_message_field() { - let mut data = json!({ - "messages": [ - {"messageId": "m1", "subject": "Hello"}, - {"messageId": "m2", "subject": "World"}, - ] - }); - let top_md = "## Hello\nbody A — link https://gh.io/abc\n---\n## World\nbody B"; - super::apply_response_level_markdown(&mut data, top_md); - let m1 = data["messages"][0]["markdownFormatted"].as_str().unwrap(); - let m2 = data["messages"][1]["markdownFormatted"].as_str().unwrap(); - assert!(m1.contains("Hello")); - assert!( - m1.contains("https://gh.io/abc"), - "shortened URL must survive" - ); - assert!(m2.contains("World")); - assert!(!m1.contains("World"), "no cross-message bleed"); -} diff --git a/src/memory/sync/composio/providers/normalize/helpers.rs b/src/memory/sync/composio/providers/normalize/helpers.rs deleted file mode 100644 index 101239e..0000000 --- a/src/memory/sync/composio/providers/normalize/helpers.rs +++ /dev/null @@ -1,50 +0,0 @@ -//! Shared helpers for the provider normalisers in this module. - -/// Walk a JSON object using a list of dotted-path candidates and return the -/// first non-empty **string** match. -/// -/// # This is deliberately NOT `super::super::common::pick_str` -/// -/// The crate carries two `pick_str` functions with the same name and -/// genuinely different behaviour. Do not "deduplicate" them: -/// -/// | | this one (`normalize::helpers`) | `common::pick_str` | -/// |---|---|---| -/// | traversal | `Value::get` per `.`-separated segment — objects only | `Value::pointer` — also indexes into arrays | -/// | non-string leaf | rejected, returns `None` | `Number` is coerced via `to_string()` | -/// -/// The number case is the one that bites. A payload whose `id` is `42` -/// rather than `"42"` yields `None` here and `Some("42")` there, which -/// silently changes what a normaliser emits as a document id. The callers of -/// this function were written against the reject-non-strings behaviour and -/// have a test pinning it (`pick_str_rejects_non_string_values` below, and -/// the host-side mirror of it). -pub fn pick_str(value: &serde_json::Value, paths: &[&str]) -> Option { - for path in paths { - let mut cur = value; - let mut ok = true; - for segment in path.split('.') { - match cur.get(segment) { - Some(next) => cur = next, - None => { - ok = false; - break; - } - } - } - if !ok { - continue; - } - if let Some(s) = cur.as_str() { - let trimmed = s.trim(); - if !trimmed.is_empty() { - return Some(trimmed.to_string()); - } - } - } - None -} - -#[cfg(test)] -#[path = "helpers_tests.rs"] -mod tests; diff --git a/src/memory/sync/composio/providers/normalize/helpers_tests.rs b/src/memory/sync/composio/providers/normalize/helpers_tests.rs deleted file mode 100644 index e481f19..0000000 --- a/src/memory/sync/composio/providers/normalize/helpers_tests.rs +++ /dev/null @@ -1,35 +0,0 @@ -use super::*; -use serde_json::json; - -#[test] -fn pick_str_finds_first_non_empty_match() { - let v = json!({"data": {"user": {"name": "Ada", "email": "ada@example.com"}}}); - assert_eq!( - pick_str(&v, &["data.user.name", "data.user.email"]), - Some("Ada".into()) - ); - assert_eq!( - pick_str(&v, &["data.missing", "data.user.email"]), - Some("ada@example.com".into()) - ); - assert_eq!(pick_str(&v, &["nope.nope"]), None); -} - -#[test] -fn pick_str_respects_path_order() { - let v = json!({"a": "first", "b": "second"}); - assert_eq!(pick_str(&v, &["a", "b"]), Some("first".into())); - assert_eq!(pick_str(&v, &["b", "a"]), Some("second".into())); -} - -/// The drift guard for the divergence documented on [`pick_str`]. If this -/// ever starts returning `Some("42")`, someone has re-pointed the -/// normalisers at `common::pick_str` and changed their output. -#[test] -fn pick_str_rejects_non_string_values() { - let v = json!({"count": 42, "flag": true, "empty": "", "whitespace": " "}); - assert_eq!(pick_str(&v, &["count"]), None); - assert_eq!(pick_str(&v, &["flag"]), None); - assert_eq!(pick_str(&v, &["empty"]), None); - assert_eq!(pick_str(&v, &["whitespace"]), None); -} diff --git a/src/memory/sync/composio/providers/normalize/linear.rs b/src/memory/sync/composio/providers/normalize/linear.rs deleted file mode 100644 index 843f98a..0000000 --- a/src/memory/sync/composio/providers/normalize/linear.rs +++ /dev/null @@ -1,157 +0,0 @@ -//! Linear host normalization helpers — result extraction, issue-title extraction, -//! viewer identity, cursor extraction, and time utilities. -//! -//! Linear's GraphQL API (and therefore Composio's wrapping of it) returns -//! connection-style lists (`{ nodes: [...], pageInfo: {...} }`) at the top -//! level or nested under `data`. The functions here walk the union of -//! common shapes so the provider does not have to branch per Composio -//! envelope variant. - -use serde_json::Value; - -use super::helpers::pick_str; - -/// Walk the Composio response envelope for Linear issue list results. -/// -/// Linear's list endpoints return `{ nodes: [...] }` or -/// `{ issues: { nodes: [...] } }` shapes; Composio may re-wrap the -/// upstream payload under `data` or `data.data`. We probe each shape -/// in order and return the first array we find. -pub fn extract_issues(data: &Value) -> Vec { - let candidates = [ - data.pointer("/data/nodes"), - data.pointer("/nodes"), - data.pointer("/data/issues/nodes"), - data.pointer("/issues/nodes"), - data.pointer("/data/data/nodes"), - data.pointer("/data/data/issues/nodes"), - data.pointer("/data/results"), - data.pointer("/results"), - data.pointer("/data/items"), - data.pointer("/items"), - ]; - for cand in candidates.into_iter().flatten() { - if let Some(arr) = cand.as_array() { - return arr.clone(); - } - } - Vec::new() -} - -/// Extract a human-readable title from a Linear issue object. -/// -/// Linear issues store the name at `title` (or `data.title` after -/// Composio envelope wrapping). Falls back to `name` / `identifier` -/// so the chunk remains identifiable even for unusual response shapes. -pub fn extract_issue_title(issue: &Value) -> Option { - pick_str( - issue, - &[ - "title", - "data.title", - "name", - "data.name", - "identifier", - "data.identifier", - ], - ) -} - -/// Extract a stable cursor timestamp from a Linear issue object. -/// -/// Linear uses ISO-8601 strings for timestamps (`updatedAt`). We keep -/// the value as a string so lexicographic comparison against the stored -/// cursor is valid. -pub fn extract_issue_updated(issue: &Value) -> Option { - pick_str( - issue, - &[ - "updatedAt", - "data.updatedAt", - "updated_at", - "data.updated_at", - ], - ) -} - -/// Extract the viewer (authenticated user) object from a -/// `LINEAR_LIST_LINEAR_USERS { isMe: true }` response. -/// -/// Linear's GraphQL viewer endpoint returns `{ nodes: [{ id, email, … }] }`. -/// Composio may wrap this under `data` or `data.data`. We probe each -/// shape and return the first element of the nodes array, falling back -/// to the payload itself if it looks like a direct user object (has -/// `id` or `email`). -pub fn extract_viewer(data: &Value) -> Option { - let array_candidates = [ - data.pointer("/data/nodes"), - data.pointer("/nodes"), - data.pointer("/data/data/nodes"), - data.pointer("/data/users/nodes"), - ]; - for cand in array_candidates.into_iter().flatten() { - if let Some(arr) = cand.as_array() { - if let Some(first) = arr.first() { - return Some(first.clone()); - } - } - } - // Fallback: if the payload itself looks like a user object, return it. - if data.get("id").is_some() || data.get("email").is_some() { - return Some(data.clone()); - } - None -} - -/// Extract the viewer's ID string from a `LINEAR_LIST_LINEAR_USERS` -/// response. Returns `None` if the payload does not contain a -/// recognizable user ID. -pub fn extract_viewer_id(data: &Value) -> Option { - let viewer = extract_viewer(data)?; - pick_str(&viewer, &["id", "data.id"]) -} - -/// Extract a pagination cursor from a Linear connection `pageInfo` block. -/// -/// Returns `Some(endCursor)` only when `hasNextPage` is `true`; -/// `None` when the last page has been reached or when the envelope does -/// not carry `pageInfo` at all. -pub fn extract_pagination_cursor(data: &Value) -> Option { - // Mirrors the `extract_issues` envelope shapes, so every shape that can - // carry a node list can also carry its `pageInfo` cursor. - let page_info_candidates = [ - data.pointer("/data/pageInfo"), - data.pointer("/pageInfo"), - data.pointer("/data/data/pageInfo"), - data.pointer("/data/issues/pageInfo"), - data.pointer("/data/data/issues/pageInfo"), - ]; - for cand in page_info_candidates.into_iter().flatten() { - let has_next = cand - .get("hasNextPage") - .and_then(|v| v.as_bool()) - .unwrap_or(false); - if has_next { - if let Some(cursor) = cand.get("endCursor").and_then(|v| v.as_str()) { - let trimmed = cursor.trim(); - if !trimmed.is_empty() { - return Some(trimmed.to_string()); - } - } - } - } - None -} - -/// Current wall-clock time in milliseconds since the UNIX epoch. -pub fn now_ms() -> u64 { - use std::time::{SystemTime, UNIX_EPOCH}; - SystemTime::now() - .duration_since(UNIX_EPOCH) - .map(|d| d.as_millis() as u64) - .unwrap_or(0) -} - -#[cfg(test)] -#[path = "linear_tests.rs"] -mod tests; diff --git a/src/memory/sync/composio/providers/normalize/linear_tests.rs b/src/memory/sync/composio/providers/normalize/linear_tests.rs deleted file mode 100644 index d796f67..0000000 --- a/src/memory/sync/composio/providers/normalize/linear_tests.rs +++ /dev/null @@ -1,184 +0,0 @@ -use super::*; -use serde_json::json; - -// ── extract_issues ─────────────────────────────────────────────── - -#[test] -fn extract_issues_from_data_nodes() { - let data = json!({ "data": { "nodes": [{"id": "i1"}, {"id": "i2"}] } }); - assert_eq!(extract_issues(&data).len(), 2); -} - -#[test] -fn extract_issues_from_top_level_nodes() { - let data = json!({ "nodes": [{"id": "i3"}] }); - assert_eq!(extract_issues(&data).len(), 1); -} - -#[test] -fn extract_issues_from_data_issues_nodes() { - let data = - json!({ "data": { "issues": { "nodes": [{"id": "i4"}, {"id": "i5"}, {"id": "i6"}] } } }); - assert_eq!(extract_issues(&data).len(), 3); -} - -#[test] -fn extract_issues_from_top_level_issues_nodes() { - let data = json!({ "issues": { "nodes": [{"id": "i7"}] } }); - assert_eq!(extract_issues(&data).len(), 1); -} - -#[test] -fn extract_issues_from_doubly_nested_issues_nodes() { - let data = - json!({ "data": { "data": { "issues": { "nodes": [{"id": "i8"}, {"id": "i9"}] } } } }); - assert_eq!(extract_issues(&data).len(), 2); -} - -#[test] -fn extract_issues_from_results() { - let data = json!({ "results": [{"id": "i7"}] }); - assert_eq!(extract_issues(&data).len(), 1); -} - -#[test] -fn extract_issues_empty_when_missing() { - let data = json!({ "foo": "bar" }); - assert!(extract_issues(&data).is_empty()); -} - -// ── extract_issue_title ────────────────────────────────────────── - -#[test] -fn extract_issue_title_from_title_field() { - let issue = json!({ "id": "i1", "title": "Fix the login bug" }); - assert_eq!( - extract_issue_title(&issue), - Some("Fix the login bug".into()) - ); -} - -#[test] -fn extract_issue_title_falls_back_to_wrapped_data() { - let issue = json!({ "data": { "title": "Wrapped issue" } }); - assert_eq!(extract_issue_title(&issue), Some("Wrapped issue".into())); -} - -#[test] -fn extract_issue_title_falls_back_to_identifier() { - let issue = json!({ "identifier": "ENG-42" }); - assert_eq!(extract_issue_title(&issue), Some("ENG-42".into())); -} - -// ── extract_issue_updated ──────────────────────────────────────── - -#[test] -fn extract_issue_updated_from_updated_at() { - let issue = json!({ "updatedAt": "2026-03-01T12:00:00.000Z" }); - assert_eq!( - extract_issue_updated(&issue), - Some("2026-03-01T12:00:00.000Z".to_string()) - ); -} - -#[test] -fn extract_issue_updated_falls_back_to_snake_case() { - let issue = json!({ "data": { "updated_at": "2026-01-15T08:30:00.000Z" } }); - assert_eq!( - extract_issue_updated(&issue), - Some("2026-01-15T08:30:00.000Z".to_string()) - ); -} - -// ── extract_viewer ─────────────────────────────────────────────── - -#[test] -fn extract_viewer_from_data_nodes() { - let data = json!({ "data": { "nodes": [{ "id": "usr_1", "email": "a@b.com" }] } }); - let v = extract_viewer(&data).expect("should find viewer"); - assert_eq!(v["id"], "usr_1"); -} - -#[test] -fn extract_viewer_from_top_level_nodes() { - let data = json!({ "nodes": [{ "id": "usr_2" }] }); - let v = extract_viewer(&data).expect("should find viewer"); - assert_eq!(v["id"], "usr_2"); -} - -#[test] -fn extract_viewer_fallback_direct_object() { - let data = json!({ "id": "usr_direct", "name": "Direct User" }); - let v = extract_viewer(&data).expect("should return direct object"); - assert_eq!(v["id"], "usr_direct"); -} - -#[test] -fn extract_viewer_returns_none_when_absent() { - let data = json!({ "foo": "bar" }); - assert!(extract_viewer(&data).is_none()); -} - -// ── extract_pagination_cursor ──────────────────────────────────── - -#[test] -fn extract_pagination_cursor_returns_cursor_when_has_next_page() { - let data = json!({ - "data": { - "pageInfo": { - "hasNextPage": true, - "endCursor": "cursor_abc" - } - } - }); - assert_eq!( - extract_pagination_cursor(&data), - Some("cursor_abc".to_string()) - ); -} - -#[test] -fn extract_pagination_cursor_returns_none_when_last_page() { - let data = json!({ - "pageInfo": { - "hasNextPage": false, - "endCursor": "cursor_xyz" - } - }); - assert!(extract_pagination_cursor(&data).is_none()); -} - -#[test] -fn extract_pagination_cursor_from_doubly_nested_issues() { - // The same `data.data.issues` shape `extract_issues` reads must also - // expose its pageInfo cursor, or a doubly-nested payload never pages. - let data = json!({ - "data": { - "data": { - "issues": { - "pageInfo": { - "hasNextPage": true, - "endCursor": "cursor_issue_2" - } - } - } - } - }); - assert_eq!( - extract_pagination_cursor(&data), - Some("cursor_issue_2".to_string()) - ); -} - -#[test] -fn extract_pagination_cursor_returns_none_when_absent() { - let data = json!({ "nodes": [{"id": "i1"}] }); - assert!(extract_pagination_cursor(&data).is_none()); -} - -// ── now_ms ─────────────────────────────────────────────────────── - -#[test] -fn now_ms_returns_nonzero() { - assert!(now_ms() > 0); -} diff --git a/src/memory/sync/composio/providers/normalize/mod.rs b/src/memory/sync/composio/providers/normalize/mod.rs deleted file mode 100644 index 5f90652..0000000 --- a/src/memory/sync/composio/providers/normalize/mod.rs +++ /dev/null @@ -1,23 +0,0 @@ -//! Provider payload normalisers for hosts that drive Composio through their -//! own provider abstraction, rather than through the [`SyncPipeline`] -//! implementations in this directory's siblings. -//! -//! These are pure `serde_json::Value` → `Value` transforms: given a raw -//! Composio action response, pull out the fields that make up a task, an -//! issue, a page or a message. They hold no credentials, touch no network, -//! and make no scheduling decisions — provider-specific normalisation is -//! driver-side by definition (see the host's `docs/specs/kernel.md` §4). -//! -//! [`SyncPipeline`]: crate::memory::sync::traits::SyncPipeline - -pub mod clickup; -pub mod github; -pub mod helpers; -pub mod linear; -pub mod notion; - -// Named `_post_process` rather than ``: `slack.rs` and -// `github.rs` (the SyncPipeline implementations) already occupy those names one -// directory up, and `gmail.rs` one directory above that. -pub mod gmail_post_process; -pub mod slack_post_process; diff --git a/src/memory/sync/composio/providers/normalize/notion.rs b/src/memory/sync/composio/providers/normalize/notion.rs deleted file mode 100644 index 0f73670..0000000 --- a/src/memory/sync/composio/providers/normalize/notion.rs +++ /dev/null @@ -1,115 +0,0 @@ -//! Notion host normalization helpers — result extraction, pagination cursor, -//! page title extraction, and time utilities. - -use serde_json::Value; - -use super::helpers::pick_str; - -/// Walk the Composio response envelope for Notion page results. -pub fn extract_results(data: &Value) -> Vec { - let candidates = [ - data.pointer("/data/results"), - data.pointer("/results"), - data.pointer("/data/data/results"), - data.pointer("/data/items"), - data.pointer("/items"), - ]; - for cand in candidates.into_iter().flatten() { - if let Some(arr) = cand.as_array() { - return arr.clone(); - } - } - Vec::new() -} - -/// Extract the rendered page body markdown from a `NOTION_GET_PAGE_MARKDOWN` -/// response. Composio wraps action output in varying envelope shapes, so we -/// try the common locations tolerantly and return the first non-empty string. -/// Returns `None` if no markdown field is found (caller falls back to the -/// metadata-only body and logs the raw shape for diagnosis). -pub fn extract_page_markdown(data: &Value) -> Option { - const PATHS: &[&str] = &[ - "/markdown", - "/data/markdown", - "/data/response_data/markdown", - "/response_data/markdown", - "/data/content", - "/content", - "/data/markdown_content", - "/markdown_content", - "/text", - "/data/text", - ]; - for p in PATHS { - if let Some(s) = data.pointer(p).and_then(Value::as_str) { - if !s.trim().is_empty() { - return Some(s.to_string()); - } - } - } - None -} - -/// Extract the Notion pagination cursor (for `start_cursor` on the -/// next request). -pub fn extract_notion_cursor(data: &Value) -> Option { - let candidates = [ - data.pointer("/data/next_cursor"), - data.pointer("/next_cursor"), - data.pointer("/data/data/next_cursor"), - ]; - for cand in candidates.into_iter().flatten() { - if let Some(s) = cand.as_str() { - let trimmed = s.trim(); - if !trimmed.is_empty() { - return Some(trimmed.to_string()); - } - } - } - None -} - -/// Try to extract a human-readable title from a Notion page object. -/// -/// Notion pages store the title in `properties.title` or -/// `properties.Name.title[0].plain_text`. We try several shapes. -pub fn extract_page_title(page: &Value) -> Option { - // Try the common `properties.title.title[0].plain_text` shape. - let props = page - .get("properties") - .or_else(|| page.get("data")?.get("properties")); - if let Some(props) = props { - // Walk all properties looking for a "title" type field. - if let Some(obj) = props.as_object() { - for (_key, val) in obj { - if val.get("type").and_then(Value::as_str) == Some("title") { - if let Some(arr) = val.get("title").and_then(Value::as_array) { - let text: String = arr - .iter() - .filter_map(|t| t.get("plain_text").and_then(Value::as_str)) - .collect::>() - .join(""); - if !text.is_empty() { - return Some(text); - } - } - } - } - } - } - - // Fallback: top-level "title" field (some Composio shapes). - pick_str(page, &["title", "data.title", "name", "data.name"]) -} - -pub fn now_ms() -> u64 { - use std::time::{SystemTime, UNIX_EPOCH}; - SystemTime::now() - .duration_since(UNIX_EPOCH) - .map(|d| d.as_millis() as u64) - .unwrap_or(0) -} - -#[cfg(test)] -#[path = "notion_tests.rs"] -mod tests; diff --git a/src/memory/sync/composio/providers/normalize/notion_tests.rs b/src/memory/sync/composio/providers/normalize/notion_tests.rs deleted file mode 100644 index f235b3a..0000000 --- a/src/memory/sync/composio/providers/normalize/notion_tests.rs +++ /dev/null @@ -1,137 +0,0 @@ -use super::*; -use serde_json::json; - -#[test] -fn extract_results_from_data_results() { - let data = json!({"data": {"results": [{"id": "page1"}]}}); - let results = extract_results(&data); - assert_eq!(results.len(), 1); -} - -#[test] -fn extract_page_markdown_reads_top_level_field() { - // Matches the live GET_PAGE_MARKDOWN envelope observed empirically: - // {id, markdown, object, request_id, truncated, unknown_block_ids}. - let data = json!({ - "id": "p1", - "markdown": "# Heading\n\nbody text", - "object": "page", - "truncated": false, - }); - assert_eq!( - extract_page_markdown(&data).as_deref(), - Some("# Heading\n\nbody text") - ); -} - -#[test] -fn extract_page_markdown_reads_nested_envelope() { - let data = json!({ "data": { "markdown": "nested body" } }); - assert_eq!(extract_page_markdown(&data).as_deref(), Some("nested body")); -} - -#[test] -fn extract_page_markdown_none_for_empty_or_missing() { - // Empty markdown (a DB row with no body blocks) → None → metadata-only. - assert_eq!(extract_page_markdown(&json!({ "markdown": "" })), None); - assert_eq!(extract_page_markdown(&json!({ "markdown": " " })), None); - // No markdown field at all → None. - assert_eq!(extract_page_markdown(&json!({ "id": "p1" })), None); -} - -#[test] -fn extract_results_from_top_level() { - let data = json!({"results": [{"id": "a"}, {"id": "b"}]}); - let results = extract_results(&data); - assert_eq!(results.len(), 2); -} - -#[test] -fn extract_results_from_data_items() { - let data = json!({"data": {"items": [{"id": "x"}]}}); - let results = extract_results(&data); - assert_eq!(results.len(), 1); -} - -#[test] -fn extract_results_empty_when_no_match() { - let data = json!({"foo": "bar"}); - assert!(extract_results(&data).is_empty()); -} - -#[test] -fn extract_notion_cursor_from_data() { - let data = json!({"data": {"next_cursor": "cur123"}}); - assert_eq!(extract_notion_cursor(&data), Some("cur123".into())); -} - -#[test] -fn extract_notion_cursor_from_top_level() { - let data = json!({"next_cursor": "abc"}); - assert_eq!(extract_notion_cursor(&data), Some("abc".into())); -} - -#[test] -fn extract_notion_cursor_none_when_empty() { - let data = json!({"data": {"next_cursor": " "}}); - assert_eq!(extract_notion_cursor(&data), None); -} - -#[test] -fn extract_notion_cursor_none_when_missing() { - assert_eq!(extract_notion_cursor(&json!({})), None); -} - -#[test] -fn extract_page_title_from_properties_title_type() { - let page = json!({ - "properties": { - "Name": { - "type": "title", - "title": [{"plain_text": "Hello"}, {"plain_text": " World"}] - } - } - }); - assert_eq!(extract_page_title(&page), Some("Hello World".into())); -} - -#[test] -fn extract_page_title_from_nested_data_properties() { - let page = json!({ - "data": { - "properties": { - "Title": { - "type": "title", - "title": [{"plain_text": "My Page"}] - } - } - } - }); - assert_eq!(extract_page_title(&page), Some("My Page".into())); -} - -#[test] -fn extract_page_title_fallback_to_top_level_title() { - let page = json!({"title": "Fallback Title"}); - assert_eq!(extract_page_title(&page), Some("Fallback Title".into())); -} - -#[test] -fn extract_page_title_none_when_empty() { - let page = json!({"properties": {"Name": {"type": "title", "title": []}}}); - // Empty title array means no text - assert!( - extract_page_title(&page).is_none() || extract_page_title(&page) == Some(String::new()) - ); -} - -#[test] -fn extract_page_title_none_when_no_title_field() { - let page = json!({"id": "123"}); - assert!(extract_page_title(&page).is_none()); -} - -#[test] -fn now_ms_returns_nonzero() { - assert!(now_ms() > 0); -} diff --git a/src/memory/sync/composio/providers/normalize/slack_post_process.rs b/src/memory/sync/composio/providers/normalize/slack_post_process.rs deleted file mode 100644 index 6ada3e1..0000000 --- a/src/memory/sync/composio/providers/normalize/slack_post_process.rs +++ /dev/null @@ -1,320 +0,0 @@ -//! Slack-specific post-processing of Composio action responses. -//! -//! Composio's Slack responses are verbose API envelopes. This module -//! rewrites each supported action's response into a slim, stable shape -//! that the ingest pipeline and enrichers can consume without walking -//! Composio's unstable nested envelopes. -//! -//! ## Supported slugs -//! -//! - `SLACK_FETCH_CONVERSATION_HISTORY` — reshapes into top-level -//! `messages[]` with `{ ts, user, text, thread_ts, channel_id }`. -//! Empty-text messages are dropped. `channel_id` is absent here (it's -//! in the request, not the response); the caller injects it via the -//! enricher in -//! [`crate::memory::sync::composio::providers::SlackSyncPipeline`]. -//! -//! - `SLACK_LIST_CONVERSATIONS` — reshapes into top-level `channels[]` -//! with `{ id, name, is_private }` per channel. Entries with an empty -//! id are dropped. -//! -//! - `SLACK_SEARCH_MESSAGES` — reshapes `messages.matches[]` (possibly -//! nested) into top-level `messages[]` with `{ ts, user, text, -//! thread_ts, channel_id }`. `channel_id` is pulled from each match's -//! `channel.id` field. `paging.pages` is preserved at top-level for -//! caller pagination. -//! -//! ## Design note: user-id resolution is NOT here -//! -//! `SlackUsers` is a per-sync cache built from a separate API call — -//! not a function of any individual response. Resolving user ids -//! happens in -//! [`crate::memory::sync::composio::providers::SlackSyncPipeline`] -//! (the enricher layer), keeping this module purely data-shape–oriented. -//! This matches Gmail's pattern of "post_process is data-only". -//! -//! Unknown slugs are silently no-ops so new Composio actions don't -//! break the provider. - -use serde_json::{Map, Value}; - -/// Entry point called from `SlackProvider::post_process_action_result`. -/// -/// Dispatches on the Composio action slug and rewrites `data` in place. -/// Unknown slugs are silently ignored. -pub fn post_process(slug: &str, _arguments: Option<&Value>, data: &mut Value) { - log::debug!("[composio:slack][post-process] slug={slug}"); - match slug { - "SLACK_FETCH_CONVERSATION_HISTORY" => reshape_fetch_history(data), - "SLACK_LIST_CONVERSATIONS" => reshape_list_conversations(data), - "SLACK_SEARCH_MESSAGES" => reshape_search_messages(data), - _ => { - log::debug!("[composio:slack][post-process] unknown slug={slug}, passing through"); - } - } -} - -// ─── SLACK_FETCH_CONVERSATION_HISTORY ────────────────────────────────────── - -/// Rewrite a `SLACK_FETCH_CONVERSATION_HISTORY` response in place. -/// -/// Walks possible nested envelopes (`/data/messages`, `/messages`, -/// `/data/data/messages`) to find the raw messages array, drops messages -/// with empty `text`, and emits a slim `{ ts, user, text, thread_ts }` -/// shape under a top-level `messages[]` key. The consumed nested array is -/// removed from the payload so the raw verbose rows don't linger alongside -/// the slim copy. The caller injects `channel_id` via -/// [`super::sync::extract_messages`]. -fn reshape_fetch_history(data: &mut Value) { - let arr = take_array( - data, - &["/data/messages", "/messages", "/data/data/messages"], - 0, - ); - let slim: Vec = arr.into_iter().filter_map(slim_history_message).collect(); - let obj = ensure_object(data); - obj.insert("messages".to_string(), Value::Array(slim)); - log::debug!("[composio:slack][post-process] SLACK_FETCH_CONVERSATION_HISTORY reshaped"); -} - -fn slim_history_message(raw: Value) -> Option { - let text = raw - .get("text") - .and_then(|v| v.as_str()) - .unwrap_or("") - .trim(); - if text.is_empty() { - return None; - } - let mut out = Map::new(); - if let Some(ts) = raw.get("ts") { - out.insert("ts".into(), ts.clone()); - } else { - return None; // ts is required — no ts means we can't cursor or archive - } - if let Some(user) = raw.get("user").or_else(|| raw.get("bot_id")) { - out.insert("user".into(), user.clone()); - } - out.insert("text".into(), Value::String(text.to_string())); - if let Some(thread_ts) = raw.get("thread_ts") { - out.insert("thread_ts".into(), thread_ts.clone()); - } - if let Some(permalink) = raw.get("permalink") { - out.insert("permalink".into(), permalink.clone()); - } - Some(Value::Object(out)) -} - -/// Find the first array at any of `candidates`, remove that field (plus -/// `envelope_depth` ancestor object envelopes) from `data`, and return the -/// array. Removing the consumed nested payload keeps the reshaped output from -/// carrying duplicate raw rows. -fn take_array(data: &mut Value, candidates: &[&str], envelope_depth: usize) -> Vec { - for path in candidates { - let arr = match data.pointer(path).and_then(|v| v.as_array().cloned()) { - Some(a) => a, - None => continue, - }; - let mut remove_path = path.to_string(); - for _ in 0..envelope_depth { - remove_path = match remove_path.rsplit_once('/') { - Some((parent, _)) => parent.to_string(), - None => break, - }; - } - remove_nested(data, &remove_path); - return arr; - } - Vec::new() -} - -/// Remove the field at `path` from `data`, pruning any ancestor object that -/// the removal left empty so a consumed `data` envelope disappears entirely -/// instead of lingering as `{}`. -fn remove_nested(data: &mut Value, path: &str) { - let segments: Vec<&str> = path - .trim_start_matches('/') - .split('/') - .filter(|s| !s.is_empty()) - .collect(); - if segments.is_empty() { - return; - } - - // Remove the leaf field. - let mut current = &mut *data; - for seg in &segments[..segments.len() - 1] { - current = match current.get_mut(*seg) { - Some(next) => next, - None => return, - }; - } - if let Value::Object(map) = current { - map.remove(segments[segments.len() - 1]); - } - - // Prune empty object ancestors, deepest first. - for depth in (0..segments.len().saturating_sub(1)).rev() { - // Re-walk to the object at `segments[..=depth]`. - let mut ancestor = &mut *data; - for seg in &segments[..=depth] { - ancestor = match ancestor.get_mut(*seg) { - Some(next) => next, - None => return, - }; - } - if !matches!(ancestor, Value::Object(m) if m.is_empty()) { - break; - } - // Remove it from its parent (`segments[..depth]`). For `depth == 0` - // the parent is the top-level object, so an emptied `data` envelope - // key disappears entirely. - let mut parent = &mut *data; - for seg in &segments[..depth] { - parent = match parent.get_mut(*seg) { - Some(next) => next, - None => return, - }; - } - if let Value::Object(map) = parent { - map.remove(segments[depth]); - } - } -} - -// ─── SLACK_LIST_CONVERSATIONS ─────────────────────────────────────────────── - -/// Rewrite a `SLACK_LIST_CONVERSATIONS` response in place. -/// -/// Reshapes into a top-level `channels[]` with `{ id, name, is_private }` -/// per channel; entries with an empty id are dropped. -fn reshape_list_conversations(data: &mut Value) { - let arr = take_array( - data, - &[ - "/data/channels", - "/channels", - "/data/data/channels", - "/data/conversations", - "/conversations", - ], - 0, - ); - - let slim: Vec = arr.into_iter().filter_map(slim_channel).collect(); - let obj = ensure_object(data); - obj.insert("channels".to_string(), Value::Array(slim)); - log::debug!("[composio:slack][post-process] SLACK_LIST_CONVERSATIONS reshaped"); -} - -fn slim_channel(raw: Value) -> Option { - let id = raw.get("id").and_then(|v| v.as_str()).unwrap_or("").trim(); - if id.is_empty() { - return None; - } - let name = raw - .get("name") - .and_then(|v| v.as_str()) - .unwrap_or(id) - .trim(); - let is_private = raw - .get("is_private") - .and_then(|v| v.as_bool()) - .unwrap_or(false); - Some(Value::Object({ - let mut m = Map::new(); - m.insert("id".into(), Value::String(id.to_string())); - m.insert("name".into(), Value::String(name.to_string())); - m.insert("is_private".into(), Value::Bool(is_private)); - m - })) -} - -// ─── SLACK_SEARCH_MESSAGES ────────────────────────────────────────────────── - -/// Rewrite a `SLACK_SEARCH_MESSAGES` response in place. -/// -/// Reshapes `messages.matches[]` (possibly nested under one or two -/// `data` envelopes) into top-level `messages[]`. `channel_id` is pulled -/// from each match's `channel.id` field. `paging.pages` is preserved at -/// top-level under `pages` for the caller to drive pagination. -fn reshape_search_messages(data: &mut Value) { - // Preserve paging info before mutating data (take_array below removes the - // envelope that carries it). - let pages = [ - data.pointer("/data/messages/paging/pages"), - data.pointer("/messages/paging/pages"), - data.pointer("/data/data/messages/paging/pages"), - ] - .into_iter() - .flatten() - .find_map(|v| v.as_u64()) - .unwrap_or(1); - - // Envelope depth 1 removes the `messages` object (matches + paging) that - // held the consumed rows, not just the `matches` array. - let arr = take_array( - data, - &[ - "/data/messages/matches", - "/messages/matches", - "/data/data/messages/matches", - ], - 1, - ); - - let slim: Vec = arr.into_iter().filter_map(slim_search_match).collect(); - let obj = ensure_object(data); - obj.insert("messages".to_string(), Value::Array(slim)); - obj.insert("pages".to_string(), Value::Number(pages.into())); - log::debug!("[composio:slack][post-process] SLACK_SEARCH_MESSAGES reshaped"); -} - -fn slim_search_match(raw: Value) -> Option { - let text = raw - .get("text") - .and_then(|v| v.as_str()) - .unwrap_or("") - .trim(); - if text.is_empty() { - return None; - } - let ts = raw.get("ts")?; - let channel_id = raw - .pointer("/channel/id") - .and_then(|v| v.as_str()) - .unwrap_or("") - .trim(); - - let mut out = Map::new(); - out.insert("ts".into(), ts.clone()); - if let Some(user) = raw.get("user").or_else(|| raw.get("bot_id")) { - out.insert("user".into(), user.clone()); - } - out.insert("text".into(), Value::String(text.to_string())); - if let Some(thread_ts) = raw.get("thread_ts") { - out.insert("thread_ts".into(), thread_ts.clone()); - } - if !channel_id.is_empty() { - out.insert("channel_id".into(), Value::String(channel_id.to_string())); - } - if let Some(permalink) = raw.get("permalink") { - out.insert("permalink".into(), permalink.clone()); - } - Some(Value::Object(out)) -} - -// ─── Helpers ──────────────────────────────────────────────────────────────── - -/// Ensure `data` is a JSON object, replacing it with an empty object if -/// not. Returns a mutable ref to the inner map. -fn ensure_object(data: &mut Value) -> &mut Map { - if !data.is_object() { - *data = Value::Object(Map::new()); - } - data.as_object_mut().unwrap() -} - -#[cfg(test)] -#[path = "slack_post_process_tests.rs"] -mod tests; diff --git a/src/memory/sync/composio/providers/normalize/slack_post_process_tests.rs b/src/memory/sync/composio/providers/normalize/slack_post_process_tests.rs deleted file mode 100644 index 7a9ff19..0000000 --- a/src/memory/sync/composio/providers/normalize/slack_post_process_tests.rs +++ /dev/null @@ -1,257 +0,0 @@ -use super::*; -use serde_json::json; - -// ─── SLACK_FETCH_CONVERSATION_HISTORY ───────────────────────────────────── - -#[test] -fn history_reshapes_top_level_messages() { - let mut data = json!({ - "messages": [ - { "ts": "1714003200.000100", "user": "U1", "text": "hello" }, - { "ts": "1714003300.000200", "user": "U2", "text": "world", "thread_ts": "1714003200.0" }, - { "ts": "1714003400.000300", "user": "U3", "text": " " }, // dropped: empty text - ], - "response_metadata": { "next_cursor": "abc" } - }); - post_process("SLACK_FETCH_CONVERSATION_HISTORY", None, &mut data); - - let msgs = data["messages"].as_array().unwrap(); - assert_eq!(msgs.len(), 2, "empty-text message must be dropped"); - assert_eq!(msgs[0]["ts"], "1714003200.000100"); - assert_eq!(msgs[0]["user"], "U1"); - assert_eq!(msgs[0]["text"], "hello"); - assert!(msgs[0].get("thread_ts").is_none()); - assert_eq!(msgs[1]["thread_ts"], "1714003200.0"); -} - -#[test] -fn history_reshapes_nested_data_envelope() { - let mut data = json!({ - "data": { - "messages": [ - { "ts": "1714003200.0", "user": "U1", "text": "hi" } - ] - } - }); - post_process("SLACK_FETCH_CONVERSATION_HISTORY", None, &mut data); - let msgs = data["messages"].as_array().unwrap(); - assert_eq!(msgs.len(), 1); - assert_eq!(msgs[0]["text"], "hi"); -} - -#[test] -fn history_reshapes_doubly_nested_envelope() { - let mut data = json!({ - "data": { - "data": { - "messages": [ - { "ts": "1714003200.0", "user": "U1", "text": "deep" } - ] - } - } - }); - post_process("SLACK_FETCH_CONVERSATION_HISTORY", None, &mut data); - let msgs = data["messages"].as_array().unwrap(); - assert_eq!(msgs.len(), 1); - assert_eq!(msgs[0]["text"], "deep"); -} - -#[test] -fn history_drops_message_without_ts() { - let mut data = json!({ - "messages": [ - { "user": "U1", "text": "no timestamp" }, - { "ts": "1714003200.0", "user": "U2", "text": "has ts" }, - ] - }); - post_process("SLACK_FETCH_CONVERSATION_HISTORY", None, &mut data); - let msgs = data["messages"].as_array().unwrap(); - assert_eq!(msgs.len(), 1); - assert_eq!(msgs[0]["text"], "has ts"); -} - -#[test] -fn history_removes_nested_envelope_after_reshape() { - let mut data = json!({ - "data": { - "messages": [ - { "ts": "1714003200.0", "user": "U1", "text": "hi" } - ] - } - }); - post_process("SLACK_FETCH_CONVERSATION_HISTORY", None, &mut data); - - let msgs = data["messages"].as_array().unwrap(); - assert_eq!(msgs.len(), 1); - assert_eq!(msgs[0]["text"], "hi"); - assert!( - data.pointer("/data").is_none(), - "consumed `data.messages` envelope must be removed, got: {data}" - ); -} - -// ─── SLACK_LIST_CONVERSATIONS ───────────────────────────────────────────── - -#[test] -fn list_conversations_reshapes_channels() { - let mut data = json!({ - "data": { - "channels": [ - { "id": "C1", "name": "eng", "is_private": false, "extra": "noise" }, - { "id": "G1", "name": "ops", "is_private": true }, - { "id": "", "name": "empty-id" }, // dropped - ] - } - }); - post_process("SLACK_LIST_CONVERSATIONS", None, &mut data); - let channels = data["channels"].as_array().unwrap(); - assert_eq!(channels.len(), 2, "empty-id entry must be dropped"); - assert_eq!(channels[0]["id"], "C1"); - assert_eq!(channels[0]["name"], "eng"); - assert_eq!(channels[0]["is_private"], false); - assert!( - channels[0].get("extra").is_none(), - "noise fields must be removed" - ); - assert_eq!(channels[1]["id"], "G1"); - assert_eq!(channels[1]["is_private"], true); -} - -#[test] -fn list_conversations_falls_back_to_conversations_key() { - let mut data = json!({ - "conversations": [ - { "id": "C2", "name": "dev", "is_private": false } - ] - }); - post_process("SLACK_LIST_CONVERSATIONS", None, &mut data); - let channels = data["channels"].as_array().unwrap(); - assert_eq!(channels.len(), 1); - assert_eq!(channels[0]["id"], "C2"); - assert!( - data.pointer("/conversations").is_none(), - "consumed `conversations` field must be removed" - ); -} - -// ─── SLACK_SEARCH_MESSAGES ──────────────────────────────────────────────── - -#[test] -fn search_messages_reshapes_matches() { - let mut data = json!({ - "messages": { - "matches": [ - { - "ts": "1714003200.0", - "user": "U1", - "text": "hello from search", - "channel": { "id": "C1" } - }, - { - "ts": "1714003300.0", - "user": "U2", - "text": " ", // dropped: whitespace only - "channel": { "id": "C1" } - }, - ], - "paging": { "pages": 3 } - } - }); - post_process("SLACK_SEARCH_MESSAGES", None, &mut data); - let msgs = data["messages"].as_array().unwrap(); - assert_eq!(msgs.len(), 1, "empty-text match must be dropped"); - assert_eq!(msgs[0]["ts"], "1714003200.0"); - assert_eq!(msgs[0]["text"], "hello from search"); - assert_eq!(msgs[0]["channel_id"], "C1"); - assert_eq!(data["pages"], 3, "paging.pages must be preserved"); -} - -#[test] -fn search_messages_nested_data_envelope() { - let mut data = json!({ - "data": { - "messages": { - "matches": [ - { "ts": "1714003200.0", "user": "U1", "text": "nested", "channel": { "id": "C2" } } - ], - "paging": { "pages": 1 } - } - } - }); - post_process("SLACK_SEARCH_MESSAGES", None, &mut data); - let msgs = data["messages"].as_array().unwrap(); - assert_eq!(msgs.len(), 1); - assert_eq!(msgs[0]["channel_id"], "C2"); - assert_eq!(data["pages"], 1_u64); -} - -#[test] -fn search_messages_no_matches_emits_empty_array() { - let mut data = json!({ "messages": { "matches": [] } }); - post_process("SLACK_SEARCH_MESSAGES", None, &mut data); - let msgs = data["messages"].as_array().unwrap(); - assert!(msgs.is_empty()); -} - -#[test] -fn search_messages_removes_nested_envelope_after_reshape() { - let mut data = json!({ - "data": { - "messages": { - "matches": [ - { "ts": "1714003200.0", "user": "U1", "text": "nested", "channel": { "id": "C2" } } - ], - "paging": { "pages": 1 } - } - } - }); - post_process("SLACK_SEARCH_MESSAGES", None, &mut data); - - let msgs = data["messages"].as_array().unwrap(); - assert_eq!(msgs.len(), 1); - assert_eq!(msgs[0]["channel_id"], "C2"); - assert_eq!(data["pages"], 1_u64); - assert!( - data.pointer("/data").is_none(), - "consumed `data.messages` envelope must be removed, got: {data}" - ); -} - -#[test] -fn search_messages_doubly_nested_paging_preserved() { - let mut data = json!({ - "data": { - "data": { - "messages": { - "matches": [ - { "ts": "1714003200.0", "user": "U1", "text": "deep", "channel": { "id": "C3" } } - ], - "paging": { "pages": 4 } - } - } - } - }); - post_process("SLACK_SEARCH_MESSAGES", None, &mut data); - - let msgs = data["messages"].as_array().unwrap(); - assert_eq!(msgs.len(), 1); - assert_eq!(msgs[0]["text"], "deep"); - assert_eq!( - data["pages"], 4_u64, - "doubly-nested paging must be preserved" - ); - assert!( - data.pointer("/data").is_none(), - "consumed `data.data.messages` envelope must be removed, got: {data}" - ); -} - -// ─── Unknown slug ───────────────────────────────────────────────────────── - -#[test] -fn unknown_slug_is_noop() { - let mut data = json!({ "foo": "bar" }); - let original = data.clone(); - post_process("SLACK_SEND_MESSAGE", None, &mut data); - assert_eq!(data, original, "unknown slug must not mutate data"); -}