From e8477985ce67b70453fee0ed9dc2589d396a0548 Mon Sep 17 00:00:00 2001 From: Shanu Date: Tue, 18 Aug 2026 12:43:41 +0530 Subject: [PATCH] Drop the host-side Composio normalisers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit tinymemory#18 §B3: "Payload normalisers are pure `Value -> Value` transforms with no engine dependency. Move them back into a `tinymemory-sync` crate — so a non-TinyCortex engine gets Composio sync for free." `providers::normalize` was fifteen files and 2,598 lines that this crate never called. Its only consumer was the host, reaching in through `tinycortex::memory::sync::composio::providers::normalize::*` — which meant a host binding a *different* memory engine could not have Composio sync at all, despite none of this code caring which engine is bound. That is the coupling §B3 names, and it ran through here. Verified dead before removing: the only references to `normalize::` outside the module itself were four doc-links in `common.rs`, plus one `use normalize::NormalizedView` in `store/safety/pii.rs` — which is unicode normalisation, an unrelated `normalize`. The four doc-links are kept and repointed. They explain that `common::pick_str` and the normalisers' `pick_str` deliberately differ — one resolves with `Value::pointer` and coerces numbers, the other walks with `Value::get` and does not — and that warning is worth more now that the two live in different crates than it was when they were siblings. The code now lives in `tinymemory-sync`, which depends on `serde_json`, two logging facades and `chrono`, and links no engine. Validation: - `cargo check --workspace --all-targets` — clean - `cargo test --workspace` — 1396 passed, 0 failed - `cargo fmt --all -- --check` — clean - `cargo clippy --all-targets -- -D warnings` (the CI gate) — clean - `RUSTDOCFLAGS="-D warnings" cargo doc --no-deps --all-features` — clean Refs tinyhumansai/tinymemory#18 (§B3) --- src/memory/sync/composio/providers/common.rs | 16 +- src/memory/sync/composio/providers/mod.rs | 1 - .../composio/providers/normalize/clickup.rs | 133 ----- .../providers/normalize/clickup_tests.rs | 96 ---- .../composio/providers/normalize/github.rs | 130 ----- .../providers/normalize/github_tests.rs | 118 ----- .../providers/normalize/gmail_post_process.rs | 489 ------------------ .../normalize/gmail_post_process_tests.rs | 354 ------------- .../composio/providers/normalize/helpers.rs | 50 -- .../providers/normalize/helpers_tests.rs | 35 -- .../composio/providers/normalize/linear.rs | 157 ------ .../providers/normalize/linear_tests.rs | 184 ------- .../sync/composio/providers/normalize/mod.rs | 23 - .../composio/providers/normalize/notion.rs | 115 ---- .../providers/normalize/notion_tests.rs | 137 ----- .../providers/normalize/slack_post_process.rs | 320 ------------ .../normalize/slack_post_process_tests.rs | 257 --------- 17 files changed, 9 insertions(+), 2606 deletions(-) delete mode 100644 src/memory/sync/composio/providers/normalize/clickup.rs delete mode 100644 src/memory/sync/composio/providers/normalize/clickup_tests.rs delete mode 100644 src/memory/sync/composio/providers/normalize/github.rs delete mode 100644 src/memory/sync/composio/providers/normalize/github_tests.rs delete mode 100644 src/memory/sync/composio/providers/normalize/gmail_post_process.rs delete mode 100644 src/memory/sync/composio/providers/normalize/gmail_post_process_tests.rs delete mode 100644 src/memory/sync/composio/providers/normalize/helpers.rs delete mode 100644 src/memory/sync/composio/providers/normalize/helpers_tests.rs delete mode 100644 src/memory/sync/composio/providers/normalize/linear.rs delete mode 100644 src/memory/sync/composio/providers/normalize/linear_tests.rs delete mode 100644 src/memory/sync/composio/providers/normalize/mod.rs delete mode 100644 src/memory/sync/composio/providers/normalize/notion.rs delete mode 100644 src/memory/sync/composio/providers/normalize/notion_tests.rs delete mode 100644 src/memory/sync/composio/providers/normalize/slack_post_process.rs delete mode 100644 src/memory/sync/composio/providers/normalize/slack_post_process_tests.rs 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"); -}